817 lines
33 KiB
Python
817 lines
33 KiB
Python
"""
|
|
Modèles pour la planification et les maintenance préventive - GMAO Collège
|
|
"""
|
|
from datetime import datetime, timezone, date, time
|
|
from ...extensions import db
|
|
|
|
|
|
class WorkSchedule(db.Model):
|
|
"""Horaires de travail."""
|
|
__tablename__ = "work_schedules"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
|
|
academic_year = db.Column(db.Integer, nullable=True, index=True)
|
|
|
|
day_of_week = db.Column(db.Integer)
|
|
start_time = db.Column(db.Time)
|
|
end_time = db.Column(db.Time)
|
|
lunch_start = db.Column(db.Time, nullable=True) # Début pause déjeuner
|
|
lunch_end = db.Column(db.Time, nullable=True) # Fin pause déjeuner
|
|
template_id = db.Column(db.Integer, db.ForeignKey("work_schedule_templates.id"), nullable=True)
|
|
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
|
|
user = db.relationship("User", backref=db.backref("work_schedules", lazy="dynamic"))
|
|
template = db.relationship("WorkScheduleTemplate", back_populates="assignments")
|
|
|
|
def __repr__(self):
|
|
return f"<WorkSchedule {self.day_of_week} {self.start_time}-{self.end_time}>"
|
|
|
|
|
|
class WorkScheduleTemplate(db.Model):
|
|
"""Un des trois profils horaires paramétrables de l'agent."""
|
|
__tablename__ = "work_schedule_templates"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False, index=True)
|
|
name = db.Column(db.String(80), nullable=False)
|
|
start_time = db.Column(db.Time, nullable=False)
|
|
end_time = db.Column(db.Time, nullable=False)
|
|
lunch_start = db.Column(db.Time, nullable=True)
|
|
lunch_end = db.Column(db.Time, nullable=True)
|
|
is_active = db.Column(db.Boolean, nullable=False, default=True)
|
|
|
|
user = db.relationship("User", backref=db.backref("work_schedule_templates", lazy="dynamic"))
|
|
assignments = db.relationship("WorkSchedule", back_populates="template")
|
|
|
|
@property
|
|
def duration_minutes(self):
|
|
total = (self.end_time.hour * 60 + self.end_time.minute) - (self.start_time.hour * 60 + self.start_time.minute)
|
|
if self.lunch_start and self.lunch_end:
|
|
total -= (self.lunch_end.hour * 60 + self.lunch_end.minute) - (self.lunch_start.hour * 60 + self.lunch_start.minute)
|
|
return max(total, 0)
|
|
|
|
|
|
class AnnualTimeConfig(db.Model):
|
|
"""Objectif annuel et déduction de pénibilité pour un agent."""
|
|
__tablename__ = "annual_time_configs"
|
|
__table_args__ = (db.UniqueConstraint("user_id", "year", name="uq_annual_time_user_year"),)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False, index=True)
|
|
year = db.Column(db.Integer, nullable=False)
|
|
target_minutes = db.Column(db.Integer, nullable=False, default=1607 * 60)
|
|
hardship_deduction_minutes = db.Column(db.Integer, nullable=False, default=0)
|
|
notes = db.Column(db.Text, nullable=True)
|
|
|
|
user = db.relationship("User", backref=db.backref("annual_time_configs", lazy="dynamic"))
|
|
|
|
@property
|
|
def minutes_to_work(self):
|
|
return max((self.target_minutes or 0) - (self.hardship_deduction_minutes or 0), 0)
|
|
|
|
|
|
class TimeEntry(db.Model):
|
|
"""Suivi réel d'une journée : retard, présence, permanence ou heures sup."""
|
|
__tablename__ = "work_time_entries"
|
|
__table_args__ = (db.UniqueConstraint("user_id", "work_date", name="uq_work_time_user_date"),)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False, index=True)
|
|
work_date = db.Column(db.Date, nullable=False, index=True)
|
|
planned_minutes = db.Column(db.Integer, nullable=False, default=0)
|
|
actual_minutes = db.Column(db.Integer, nullable=False, default=0)
|
|
entry_type = db.Column(db.String(30), nullable=False, default="travail")
|
|
note = db.Column(db.Text, nullable=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
user = db.relationship("User", backref=db.backref("work_time_entries", lazy="dynamic"))
|
|
|
|
@property
|
|
def variance_minutes(self):
|
|
return (self.actual_minutes or 0) - (self.planned_minutes or 0)
|
|
|
|
|
|
class CollegeClosure(db.Model):
|
|
"""Fermetures du collège / Vacances scolaires."""
|
|
__tablename__ = "college_closures"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
name = db.Column(db.String(255), nullable=False)
|
|
start_date = db.Column(db.Date, nullable=False)
|
|
end_date = db.Column(db.Date, nullable=False)
|
|
|
|
closure_type = db.Column(db.String(50), default='vacances')
|
|
work_hours_type = db.Column(db.String(20), default='none') # none, reduced, custom
|
|
notes = db.Column(db.Text)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
|
|
# Relation avec les horaires spécifiques
|
|
schedules = db.relationship("ClosureSchedule", backref="closure", lazy="dynamic",
|
|
cascade="all, delete-orphan")
|
|
|
|
def __repr__(self):
|
|
return f"<CollegeClosure {self.name}>"
|
|
|
|
|
|
class ClosureSchedule(db.Model):
|
|
"""Horaires de travail pendant une période de vacances."""
|
|
__tablename__ = "closure_schedules"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
closure_id = db.Column(db.Integer, db.ForeignKey("college_closures.id"), nullable=False)
|
|
|
|
day_of_week = db.Column(db.Integer, nullable=False) # 0=Dimanche, 1=Lundi, etc.
|
|
start_time = db.Column(db.Time)
|
|
end_time = db.Column(db.Time)
|
|
lunch_start = db.Column(db.Time) # Début pause déjeuner
|
|
lunch_end = db.Column(db.Time) # Fin pause déjeuner
|
|
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
|
|
def __repr__(self):
|
|
days = {0: 'Dimanche', 1: 'Lundi', 2: 'Mardi', 3: 'Mercredi', 4: 'Jeudi', 5: 'Vendredi', 6: 'Samedi'}
|
|
return f"<ClosureSchedule {days.get(self.day_of_week)} {self.start_time}-{self.end_time}>"
|
|
|
|
|
|
class ClosureWorkDay(db.Model):
|
|
"""Journée exceptionnellement travaillée pendant des vacances scolaires."""
|
|
__tablename__ = "closure_work_days"
|
|
__table_args__ = (
|
|
db.UniqueConstraint("closure_id", "work_date", name="uq_closure_work_day"),
|
|
)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
closure_id = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey("college_closures.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
template_id = db.Column(db.Integer, db.ForeignKey("work_schedule_templates.id"), nullable=True, index=True)
|
|
work_date = db.Column(db.Date, nullable=False, index=True)
|
|
start_time = db.Column(db.Time, nullable=False)
|
|
end_time = db.Column(db.Time, nullable=False)
|
|
lunch_start = db.Column(db.Time, nullable=True)
|
|
lunch_end = db.Column(db.Time, nullable=True)
|
|
notes = db.Column(db.String(255), nullable=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
|
|
closure = db.relationship(
|
|
"CollegeClosure",
|
|
backref=db.backref("work_days", cascade="all, delete-orphan", order_by="ClosureWorkDay.work_date"),
|
|
)
|
|
template = db.relationship("WorkScheduleTemplate")
|
|
|
|
def __repr__(self):
|
|
return f"<ClosureWorkDay {self.work_date} {self.start_time}-{self.end_time}>"
|
|
|
|
|
|
class PersonalLeave(db.Model):
|
|
"""Congés personnels."""
|
|
__tablename__ = "personal_leaves"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
|
|
|
|
start_date = db.Column(db.Date, nullable=False)
|
|
end_date = db.Column(db.Date, nullable=False)
|
|
leave_type = db.Column(db.String(50))
|
|
|
|
notes = db.Column(db.Text)
|
|
is_approved = db.Column(db.Boolean, default=False)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
|
|
user = db.relationship("User", backref=db.backref("personal_leaves", lazy="dynamic"))
|
|
|
|
def __repr__(self):
|
|
return f"<PersonalLeave {self.user_id} {self.start_date}-{self.end_date}>"
|
|
|
|
|
|
class Training(db.Model):
|
|
"""Formation."""
|
|
__tablename__ = "trainings"
|
|
__table_args__ = (db.UniqueConstraint("source_type", "source_id", name="uq_training_source"),)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
name = db.Column(db.String(255), nullable=False)
|
|
description = db.Column(db.Text)
|
|
|
|
start_date = db.Column(db.Date)
|
|
end_date = db.Column(db.Date)
|
|
start_time = db.Column(db.Time, nullable=True)
|
|
end_time = db.Column(db.Time, nullable=True)
|
|
location = db.Column(db.String(255))
|
|
source_type = db.Column(db.String(30), nullable=True)
|
|
source_id = db.Column(db.String(191), nullable=True)
|
|
source_subject = db.Column(db.String(500), nullable=True)
|
|
source_sender = db.Column(db.String(255), nullable=True)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
|
|
participants = db.relationship("TrainingParticipant", back_populates="training", cascade="all, delete-orphan")
|
|
|
|
def __repr__(self):
|
|
return f"<Training {self.name}>"
|
|
|
|
|
|
class TrainingParticipant(db.Model):
|
|
"""Participant à une formation."""
|
|
__tablename__ = "training_participants"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
training_id = db.Column(db.Integer, db.ForeignKey("trainings.id"))
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
|
|
|
|
is_confirmed = db.Column(db.Boolean, default=False)
|
|
notes = db.Column(db.Text)
|
|
|
|
training = db.relationship("Training", back_populates="participants")
|
|
user = db.relationship("User")
|
|
|
|
def __repr__(self):
|
|
return f"<TrainingParticipant {self.user_id}>"
|
|
|
|
|
|
class Meter(db.Model):
|
|
"""Compteur physique rattaché à un élément du patrimoine.
|
|
|
|
``equipment_id`` reste le rattachement historique, mais un compteur peut
|
|
désormais être porté directement par un bâtiment, une zone, un local ou
|
|
un logement. Les rattachements sont explicites afin de conserver
|
|
l'intégrité référentielle SQL.
|
|
"""
|
|
__tablename__ = "meters"
|
|
__table_args__ = (
|
|
db.CheckConstraint(
|
|
"(equipment_id IS NOT NULL) + (building_id IS NOT NULL) + "
|
|
"(zone_id IS NOT NULL) + (room_id IS NOT NULL) + "
|
|
"(housing_unit_id IS NOT NULL) = 1",
|
|
name="ck_meters_one_scope",
|
|
),
|
|
)
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=True, index=True)
|
|
building_id = db.Column(db.Integer, db.ForeignKey("buildings.id"), nullable=True, index=True)
|
|
zone_id = db.Column(db.Integer, db.ForeignKey("zones.id"), nullable=True, index=True)
|
|
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True, index=True)
|
|
housing_unit_id = db.Column(db.Integer, db.ForeignKey("housing_units.id"), nullable=True, index=True)
|
|
|
|
parent_id = db.Column(db.Integer, db.ForeignKey("meters.id"), nullable=True, index=True)
|
|
replaces_meter_id = db.Column(db.Integer, db.ForeignKey("meters.id"), nullable=True, index=True)
|
|
|
|
name = db.Column(db.String(100), nullable=False)
|
|
meter_type = db.Column(db.String(50), default="hours")
|
|
unit = db.Column(db.String(20), default="h")
|
|
usage = db.Column(db.String(255), nullable=True)
|
|
status = db.Column(db.String(30), nullable=False, default="active", server_default="active")
|
|
remainder_label = db.Column(db.String(255), nullable=True)
|
|
|
|
current_value = db.Column(db.Float, default=0)
|
|
initial_value = db.Column(db.Float, default=0)
|
|
|
|
warning_threshold = db.Column(db.Float)
|
|
critical_threshold = db.Column(db.Float)
|
|
|
|
maintenance_interval = db.Column(db.Float)
|
|
last_maintenance_value = db.Column(db.Float)
|
|
|
|
# Intervalle de relevé en jours (pour planification)
|
|
reading_interval_days = db.Column(db.Integer)
|
|
last_reading_date = db.Column(db.Date)
|
|
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
equipment = db.relationship("Equipment", backref=db.backref("meters", lazy="dynamic"))
|
|
building = db.relationship("Building", backref=db.backref("meters", lazy="dynamic"))
|
|
zone = db.relationship("Zone", backref=db.backref("meters", lazy="dynamic"))
|
|
room = db.relationship("Room", backref=db.backref("meters", lazy="dynamic"))
|
|
housing_unit = db.relationship("HousingUnit", backref=db.backref("meters", lazy="dynamic"))
|
|
parent = db.relationship("Meter", remote_side=[id], foreign_keys=[parent_id], backref=db.backref("children", lazy="dynamic"))
|
|
replaced_meter = db.relationship("Meter", remote_side=[id], foreign_keys=[replaces_meter_id], backref=db.backref("replacement", uselist=False))
|
|
readings = db.relationship("MeterReading", back_populates="meter", lazy="dynamic")
|
|
|
|
@property
|
|
def next_maintenance_at(self):
|
|
if self.last_maintenance_value and self.maintenance_interval:
|
|
return self.last_maintenance_value + self.maintenance_interval
|
|
return None
|
|
|
|
@property
|
|
def needs_maintenance(self):
|
|
if self.next_maintenance_at and self.current_value >= self.next_maintenance_at:
|
|
return True
|
|
if self.critical_threshold and self.current_value >= self.critical_threshold:
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def warning_level(self):
|
|
if self.critical_threshold and self.current_value >= self.critical_threshold:
|
|
return "critical"
|
|
if self.warning_threshold and self.current_value >= self.warning_threshold:
|
|
return "warning"
|
|
if self.needs_maintenance:
|
|
return "maintenance"
|
|
return "ok"
|
|
|
|
def __repr__(self):
|
|
return f"<Meter {self.name} ({self.current_value}{self.unit})>"
|
|
|
|
|
|
class MeterReading(db.Model):
|
|
"""Relevé de compteur."""
|
|
__tablename__ = "meter_readings"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
meter_id = db.Column(db.Integer, db.ForeignKey("meters.id"), nullable=False)
|
|
|
|
value = db.Column(db.Float, nullable=False)
|
|
reading_date = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
read_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"))
|
|
|
|
notes = db.Column(db.Text)
|
|
is_reset = db.Column(db.Boolean, nullable=False, default=False)
|
|
reset_reason = db.Column(db.String(255), nullable=True)
|
|
photo_filename = db.Column(db.String(255), nullable=True)
|
|
photo_path = db.Column(db.String(500), nullable=True)
|
|
|
|
meter = db.relationship("Meter", back_populates="readings")
|
|
read_by = db.relationship("User")
|
|
|
|
def __repr__(self):
|
|
return f"<MeterReading {self.value}>"
|
|
|
|
|
|
class MeterReadingCorrection(db.Model):
|
|
"""Audit immuable de chaque correction d'un relevé existant."""
|
|
__tablename__ = "meter_reading_corrections"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
reading_id = db.Column(db.Integer, db.ForeignKey("meter_readings.id", ondelete="CASCADE"), nullable=False, index=True)
|
|
old_value = db.Column(db.Float, nullable=False)
|
|
new_value = db.Column(db.Float, nullable=False)
|
|
corrected_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
|
|
corrected_at = db.Column(db.DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
|
|
reason = db.Column(db.Text, nullable=False)
|
|
|
|
reading = db.relationship("MeterReading", backref=db.backref("corrections", lazy="selectin"))
|
|
corrected_by = db.relationship("User")
|
|
|
|
|
|
class Consumable(db.Model):
|
|
"""Consommable pour maintenance."""
|
|
__tablename__ = "consumables"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
name = db.Column(db.String(200), nullable=False)
|
|
reference = db.Column(db.String(100))
|
|
category = db.Column(db.String(50))
|
|
|
|
quantity = db.Column(db.Integer, default=0)
|
|
min_quantity = db.Column(db.Integer, default=0)
|
|
unit = db.Column(db.String(20), default="pièce")
|
|
|
|
unit_price = db.Column(db.Float)
|
|
supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"))
|
|
|
|
storage_location = db.Column(db.String(100))
|
|
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
usages = db.relationship("ConsumableUsage", back_populates="consumable", lazy="dynamic")
|
|
supplier = db.relationship("Company")
|
|
|
|
@property
|
|
def is_low_stock(self):
|
|
return self.quantity <= self.min_quantity
|
|
|
|
def __repr__(self):
|
|
return f"<Consumable {self.name}>"
|
|
|
|
|
|
class ConsumableUsage(db.Model):
|
|
"""Utilisation de consommable."""
|
|
__tablename__ = "consumable_usages"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
consumable_id = db.Column(db.Integer, db.ForeignKey("consumables.id"))
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"))
|
|
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"))
|
|
|
|
quantity_used = db.Column(db.Integer, default=1)
|
|
used_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
used_by_id = db.Column(db.Integer, db.ForeignKey("users.id"))
|
|
|
|
notes = db.Column(db.Text)
|
|
|
|
consumable = db.relationship("Consumable", back_populates="usages")
|
|
equipment = db.relationship("Equipment")
|
|
intervention = db.relationship("Intervention")
|
|
used_by = db.relationship("User")
|
|
|
|
def __repr__(self):
|
|
return f"<ConsumableUsage {self.consumable_id}x{self.quantity_used}>"
|
|
|
|
|
|
class EquipmentConsumable(db.Model):
|
|
"""Consommables compatibles avec un équipement."""
|
|
__tablename__ = "equipment_consumables"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
equipment_category_id = db.Column(db.Integer, db.ForeignKey("equipment_categories.id"))
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"))
|
|
consumable_id = db.Column(db.Integer, db.ForeignKey("consumables.id"))
|
|
|
|
replacement_interval_days = db.Column(db.Integer)
|
|
replacement_interval_hours = db.Column(db.Integer)
|
|
replacement_interval_cycles = db.Column(db.Integer)
|
|
|
|
notes = db.Column(db.Text)
|
|
|
|
equipment_category = db.relationship("EquipmentCategory")
|
|
equipment = db.relationship("Equipment")
|
|
consumable = db.relationship("Consumable")
|
|
|
|
def __repr__(self):
|
|
return f"<EquipmentConsumable {self.equipment_id}-{self.consumable_id}>"
|
|
|
|
|
|
class PreventiveTask(db.Model):
|
|
"""Tâche préventive récurrente."""
|
|
__tablename__ = "preventive_tasks"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
lot_id = db.Column(db.Integer, db.ForeignKey("lots.id"))
|
|
category_id = db.Column(db.Integer, db.ForeignKey("equipment_categories.id"))
|
|
|
|
task_number = db.Column(db.String(20))
|
|
name = db.Column(db.String(255), nullable=False)
|
|
description = db.Column(db.Text)
|
|
location_pattern = db.Column(db.String(255))
|
|
|
|
period_days = db.Column(db.Integer)
|
|
duration_minutes = db.Column(db.Integer, default=30)
|
|
|
|
has_contract = db.Column(db.Boolean, default=False)
|
|
manager = db.Column(db.String(50))
|
|
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"))
|
|
contract_id = db.Column(db.Integer, db.ForeignKey("contracts.id"), nullable=True)
|
|
|
|
trigger_type = db.Column(db.String(20), default="time")
|
|
meter_type_required = db.Column(db.String(50))
|
|
meter_threshold = db.Column(db.Float)
|
|
|
|
requires_room_access = db.Column(db.Boolean, default=True)
|
|
allowed_time_slots = db.Column(db.String(50), default="always")
|
|
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
lot = db.relationship("Lot", backref=db.backref("preventive_tasks", lazy="dynamic"))
|
|
category = db.relationship("EquipmentCategory", backref=db.backref("preventive_tasks", lazy="dynamic"))
|
|
company = db.relationship("Company")
|
|
contract = db.relationship("Contract")
|
|
consumables_needed = db.relationship("PreventiveTaskConsumable", back_populates="task", lazy="dynamic")
|
|
|
|
def __repr__(self):
|
|
return f"<PreventiveTask {self.task_number}: {self.name}>"
|
|
|
|
|
|
class PreventiveTaskConsumable(db.Model):
|
|
"""Consommables nécessaires pour une tâche préventive."""
|
|
__tablename__ = "preventive_task_consumables"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
task_id = db.Column(db.Integer, db.ForeignKey("preventive_tasks.id"))
|
|
consumable_id = db.Column(db.Integer, db.ForeignKey("consumables.id"))
|
|
|
|
quantity = db.Column(db.Integer, default=1)
|
|
|
|
task = db.relationship("PreventiveTask", back_populates="consumables_needed")
|
|
consumable = db.relationship("Consumable")
|
|
|
|
def __repr__(self):
|
|
return f"<PreventiveTaskConsumable {self.task_id}-{self.consumable_id}>"
|
|
|
|
|
|
class ScheduledTask(db.Model):
|
|
"""Tâche planifiée pour un équipement."""
|
|
__tablename__ = "scheduled_tasks"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
preventive_task_id = db.Column(db.Integer, db.ForeignKey("preventive_tasks.id"))
|
|
lot_task_id = db.Column(db.Integer, db.ForeignKey("lot_tasks.id"))
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"))
|
|
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"))
|
|
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"))
|
|
contract_id = db.Column(db.Integer, db.ForeignKey("contracts.id"), nullable=True)
|
|
|
|
scheduled_date = db.Column(db.Date)
|
|
scheduled_start = db.Column(db.Time)
|
|
scheduled_end = db.Column(db.Time)
|
|
estimated_duration = db.Column(db.Integer)
|
|
|
|
status = db.Column(db.String(20), default="planned")
|
|
postpone_reason = db.Column(db.Text)
|
|
|
|
assigned_to_id = db.Column(db.Integer, db.ForeignKey("users.id"))
|
|
|
|
completed_at = db.Column(db.DateTime)
|
|
actual_duration = db.Column(db.Integer)
|
|
|
|
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"))
|
|
|
|
notes = db.Column(db.Text)
|
|
suspension_reason = db.Column(db.String(500), nullable=True)
|
|
suspended_at = db.Column(db.DateTime, nullable=True)
|
|
room_name_snapshot = db.Column(db.String(200), nullable=True)
|
|
lot_name_snapshot = db.Column(db.String(500), nullable=True)
|
|
equipment_name_snapshot = db.Column(db.String(200), nullable=True)
|
|
lifecycle_status_snapshot = db.Column(db.String(30), nullable=True)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
preventive_task = db.relationship("PreventiveTask", backref=db.backref("scheduled_tasks", lazy="dynamic"))
|
|
lot_task = db.relationship("LotTask", backref=db.backref("scheduled_tasks", lazy="dynamic"))
|
|
equipment = db.relationship("Equipment", backref=db.backref("scheduled_tasks", lazy="dynamic"))
|
|
room = db.relationship("Room")
|
|
assigned_to = db.relationship("User")
|
|
company = db.relationship("Company")
|
|
contract = db.relationship("Contract")
|
|
intervention = db.relationship("Intervention")
|
|
execution_segments = db.relationship(
|
|
"TaskExecutionSegment", back_populates="scheduled_task",
|
|
cascade="all, delete-orphan", order_by="TaskExecutionSegment.started_at",
|
|
)
|
|
|
|
@property
|
|
def is_overdue(self):
|
|
if self.status == "planned" and self.scheduled_date:
|
|
return self.scheduled_date < date.today()
|
|
return False
|
|
|
|
def can_be_ignored(self):
|
|
return self.status == 'planned'
|
|
|
|
@property
|
|
def realized_minutes(self):
|
|
"""Temps réellement saisi dans les segments d'exécution."""
|
|
return sum(int(segment.duration_minutes or 0) for segment in self.execution_segments)
|
|
|
|
@property
|
|
def remaining_minutes(self):
|
|
estimated = int(self.estimated_duration or 0)
|
|
return max(estimated - self.realized_minutes, 0) if estimated else 0
|
|
|
|
def ignore(self, reason=''):
|
|
"""Marque la tache comme ignoree."""
|
|
self.status = 'ignored'
|
|
self.postpone_reason = reason
|
|
db.session.commit()
|
|
|
|
def convert_to_intervention(self, user_id=None):
|
|
"""Convertit la tache planifiee en intervention."""
|
|
from .maintenance import Intervention
|
|
if self.intervention_id:
|
|
return None # Deja convertie
|
|
|
|
title = ''
|
|
if self.preventive_task:
|
|
title = f"[{self.scheduled_date}] {self.preventive_task.name}"[:290]
|
|
elif self.lot_task:
|
|
title = f"[{self.scheduled_date}] {self.lot_task.tache or 'Tache lot'}"[:290]
|
|
elif self.equipment:
|
|
title = f"[{self.scheduled_date}] Tache {self.equipment.name}"[:290]
|
|
else:
|
|
title = f"Tache {self.id} du {self.scheduled_date}"
|
|
|
|
interv = Intervention(
|
|
title=title,
|
|
description='',
|
|
type='preventif',
|
|
status='en_attente',
|
|
priority='normale',
|
|
equipment_id=self.equipment_id,
|
|
room_id=self.room_id,
|
|
scheduled_date=self.scheduled_date,
|
|
estimated_duration=float(self.estimated_duration or 0),
|
|
)
|
|
db.session.add(interv)
|
|
db.session.flush()
|
|
self.intervention_id = interv.id
|
|
self.status = 'converted'
|
|
db.session.commit()
|
|
return interv
|
|
|
|
def __repr__(self):
|
|
return f"<ScheduledTask {self.id}: {self.scheduled_date}>"
|
|
|
|
|
|
class TechnicianAvailability(db.Model):
|
|
"""Disponibilité du technicien."""
|
|
__tablename__ = "technician_availability"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
|
|
|
|
date = db.Column(db.Date)
|
|
start_time = db.Column(db.Time)
|
|
end_time = db.Column(db.Time)
|
|
|
|
availability_type = db.Column(db.String(20), default="work")
|
|
notes = db.Column(db.Text)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
|
|
user = db.relationship("User", backref=db.backref("availabilities", lazy="dynamic"))
|
|
|
|
def __repr__(self):
|
|
return f"<TechnicianAvailability {self.date} {self.availability_type}>"
|
|
|
|
|
|
class AdminTask(db.Model):
|
|
"""Tâche administrative récurrente."""
|
|
__tablename__ = "admin_tasks"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
# Tâche personnelle facultative ; NULL conserve la compatibilité des
|
|
# anciennes tâches administratives globales.
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True, index=True)
|
|
|
|
name = db.Column(db.String(255), nullable=False)
|
|
description = db.Column(db.Text)
|
|
|
|
frequency = db.Column(db.String(20), default="daily")
|
|
day_of_week = db.Column(db.Integer)
|
|
day_of_month = db.Column(db.Integer)
|
|
|
|
start_time = db.Column(db.Time)
|
|
duration_minutes = db.Column(db.Integer, default=30)
|
|
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
user = db.relationship("User", backref=db.backref("admin_tasks", lazy="dynamic"))
|
|
|
|
def __repr__(self):
|
|
return f"<AdminTask {self.name}>"
|
|
|
|
|
|
class ZoneAccessRule(db.Model):
|
|
"""Règle d'accès zone/salles."""
|
|
__tablename__ = "zone_access_rules"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
zone_id = db.Column(db.Integer, db.ForeignKey("zones.id"))
|
|
zone_name = db.Column(db.String(100))
|
|
|
|
blocking_room_ids = db.Column(db.Text)
|
|
|
|
blocked_hours_start = db.Column(db.Time)
|
|
blocked_hours_end = db.Column(db.Time)
|
|
|
|
notes = db.Column(db.Text)
|
|
|
|
zone = db.relationship("Zone")
|
|
|
|
|
|
class PlanningDay(db.Model):
|
|
"""Planification d'une journée."""
|
|
__tablename__ = "planning_days"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
date = db.Column(db.Date, nullable=False, unique=True)
|
|
|
|
# État de la journée
|
|
is_working_day = db.Column(db.Boolean, default=True)
|
|
is_holiday = db.Column(db.Boolean, default=False)
|
|
is_vacation = db.Column(db.Boolean, default=False)
|
|
|
|
# Notes
|
|
notes = db.Column(db.Text)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
# Relations
|
|
items = db.relationship("PlanningItem", back_populates="planning_day", cascade="all, delete-orphan")
|
|
|
|
def __repr__(self):
|
|
return f"<PlanningDay {self.date}>"
|
|
|
|
|
|
class PlanningItem(db.Model):
|
|
"""Élément planifié (intervention, formation, administrative)."""
|
|
__tablename__ = "planning_items"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
planning_day_id = db.Column(db.Integer, db.ForeignKey("planning_days.id"))
|
|
|
|
# Type d'élément
|
|
item_type = db.Column(db.String(20), nullable=False) # 'preventive', 'curative', 'admin', 'formation'
|
|
|
|
# Références optionnelles
|
|
scheduled_task_id = db.Column(db.Integer, db.ForeignKey("scheduled_tasks.id"))
|
|
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"))
|
|
admin_task_id = db.Column(db.Integer, db.ForeignKey("admin_tasks.id"))
|
|
|
|
# Données de base (copiées pour affichage rapide)
|
|
title = db.Column(db.String(255), nullable=False)
|
|
description = db.Column(db.Text)
|
|
duration_minutes = db.Column(db.Integer, default=30)
|
|
|
|
# Planification
|
|
start_time = db.Column(db.Time)
|
|
end_time = db.Column(db.Time)
|
|
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"))
|
|
|
|
# Priorité (1=haute, 5=basse)
|
|
priority = db.Column(db.Integer, default=3)
|
|
|
|
# Regroupement
|
|
group_id = db.Column(db.String(50)) # Pour grouper les tâches dans la même salle
|
|
|
|
# Présence requise (entreprise externe mais technicien présent)
|
|
requires_presence = db.Column(db.Boolean, default=False)
|
|
|
|
# État
|
|
status = db.Column(db.String(20), default="planned") # planned, in_progress, completed, postponed
|
|
postpone_reason = db.Column(db.Text)
|
|
postponed_to_date = db.Column(db.Date)
|
|
|
|
# Compteurs (pour relevés)
|
|
meter_reading_id = db.Column(db.Integer) # Référence sans FK (table créée plus tard)
|
|
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
|
|
|
|
# Relations
|
|
planning_day = db.relationship("PlanningDay", back_populates="items")
|
|
scheduled_task = db.relationship("ScheduledTask")
|
|
intervention = db.relationship("Intervention")
|
|
admin_task = db.relationship("AdminTask")
|
|
room = db.relationship("Room")
|
|
|
|
@property
|
|
def is_prioritary(self):
|
|
"""Une tâche est prioritaire si sa priorité est 1 ou 2."""
|
|
return self.priority in (1, 2)
|
|
|
|
def __repr__(self):
|
|
return f"<PlanningItem {self.item_type}: {self.title}>"
|
|
|
|
|
|
class TaskExecutionSegment(db.Model):
|
|
"""Un segment de réalisation d'une tâche planifiée.
|
|
|
|
Plusieurs segments peuvent appartenir à la même obligation. Cela permet
|
|
de mettre en pause une tonte ou un contrôle sans perdre le temps déjà
|
|
réalisé et sans créer une fausse tâche indépendante.
|
|
"""
|
|
__tablename__ = "task_execution_segments"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
scheduled_task_id = db.Column(db.Integer, db.ForeignKey("scheduled_tasks.id", ondelete="CASCADE"), nullable=False, index=True)
|
|
technician_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
|
|
started_at = db.Column(db.DateTime, nullable=False)
|
|
ended_at = db.Column(db.DateTime, nullable=True)
|
|
duration_minutes = db.Column(db.Integer, nullable=False, default=0)
|
|
status = db.Column(db.String(20), nullable=False, default="active")
|
|
pause_reason = db.Column(db.String(80), nullable=True)
|
|
note = db.Column(db.Text, nullable=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), nullable=False)
|
|
|
|
scheduled_task = db.relationship("ScheduledTask", back_populates="execution_segments")
|
|
technician = db.relationship("User")
|
|
|
|
def close(self, ended_at, reason=None):
|
|
self.ended_at = ended_at
|
|
self.status = "paused" if reason else "completed"
|
|
self.pause_reason = reason
|
|
self.duration_minutes = max(int((ended_at - self.started_at).total_seconds() // 60), 0)
|