427 lines
16 KiB
Python
427 lines
16 KiB
Python
"""
|
|
Modèle Equipment - GMAO Collège
|
|
Gestion des équipements
|
|
Compatible avec la base de données existante
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from ...extensions import db
|
|
|
|
|
|
class EquipmentCategory(db.Model):
|
|
"""Catégorie d'équipement."""
|
|
__tablename__ = "equipment_categories"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(100), nullable=False)
|
|
description = db.Column(db.Text, nullable=True)
|
|
|
|
# Relations
|
|
equipments = db.relationship("Equipment", back_populates="category", overlaps="category")
|
|
lots = db.relationship("Lot", back_populates="category")
|
|
|
|
def __repr__(self):
|
|
return f"<EquipmentCategory {self.name}>"
|
|
|
|
|
|
class Equipment(db.Model):
|
|
"""Équipement du collège."""
|
|
__tablename__ = "equipments"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(200), nullable=False)
|
|
code = db.Column(db.String(50), nullable=True)
|
|
description = db.Column(db.Text, nullable=True)
|
|
status = db.Column(db.String(20), nullable=True)
|
|
category_id = db.Column(db.Integer, db.ForeignKey("equipment_categories.id"), nullable=True)
|
|
lot_id = db.Column(db.Integer, db.ForeignKey("lots.id"), nullable=True)
|
|
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True)
|
|
parent_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=True)
|
|
is_group = db.Column(db.Boolean, default=False)
|
|
quantity = db.Column(db.Integer, default=1)
|
|
tracked_individually = db.Column(db.Boolean, default=False)
|
|
individual_number = db.Column(db.Integer, nullable=True)
|
|
position = db.Column(db.String(255), nullable=True)
|
|
position_x = db.Column(db.Float, nullable=True)
|
|
position_y = db.Column(db.Float, nullable=True)
|
|
install_date = db.Column(db.Date, nullable=True)
|
|
warranty_end = db.Column(db.Date, nullable=True)
|
|
is_deleted = db.Column(db.Boolean, default=False, nullable=False)
|
|
deleted_at = db.Column(db.DateTime, 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))
|
|
|
|
# Relations - sans backref pour éviter conflits
|
|
room = db.relationship("Room", back_populates="equipments")
|
|
category = db.relationship("EquipmentCategory", back_populates="equipments")
|
|
lot = db.relationship("Lot", backref=db.backref("equipments", lazy="dynamic"))
|
|
parent = db.relationship("Equipment", remote_side=[id], backref=db.backref("children", lazy="dynamic"))
|
|
interventions = db.relationship("Intervention", backref="equipment")
|
|
documents = db.relationship("EquipmentDocument", back_populates="equipment")
|
|
consumables = db.relationship("EquipmentConsumable", back_populates="equipment", lazy="dynamic")
|
|
room_history = db.relationship("EquipmentRoomHistory", backref="equipment", order_by="desc(EquipmentRoomHistory.changed_at)")
|
|
|
|
@property
|
|
def effective_lot_id(self):
|
|
"""Lot effectif (hérité du parent si non défini)."""
|
|
if self.lot_id:
|
|
return self.lot_id
|
|
if self.parent_id and self.parent:
|
|
return self.parent.effective_lot_id
|
|
return None
|
|
|
|
@property
|
|
def effective_category_id(self):
|
|
"""Catégorie effective (héritée du parent si non définie)."""
|
|
if self.category_id:
|
|
return self.category_id
|
|
if self.parent_id and self.parent:
|
|
return self.parent.effective_category_id
|
|
return None
|
|
|
|
@property
|
|
def effective_lot(self):
|
|
"""Lot effectif (hérité du parent si non défini)."""
|
|
lot_id = self.effective_lot_id
|
|
if lot_id:
|
|
from .maintenance import Lot
|
|
return Lot.query.get(lot_id)
|
|
return None
|
|
|
|
@property
|
|
def effective_category(self):
|
|
"""Catégorie effective (héritée du parent si non définie)."""
|
|
category_id = self.effective_category_id
|
|
if category_id:
|
|
return EquipmentCategory.query.get(category_id)
|
|
return None
|
|
|
|
@property
|
|
def all_children(self):
|
|
"""Tous les enfants (récursif)."""
|
|
children = list(self.children)
|
|
for child in self.children:
|
|
children.extend(child.all_children)
|
|
return children
|
|
|
|
@property
|
|
def individual_children(self):
|
|
"""Enfants individuels uniquement."""
|
|
return [c for c in self.all_children if not c.is_group]
|
|
|
|
@property
|
|
def full_path(self):
|
|
"""Chemin complet de localisation."""
|
|
if self.room:
|
|
return f"{self.room.building.name} / {self.room.name}" if self.room.building else self.room.name
|
|
return "—"
|
|
|
|
@property
|
|
def curative_count(self):
|
|
"""Nombre d'interventions curatives."""
|
|
return len([i for i in self.interventions if i.type == 'curatif'])
|
|
|
|
@property
|
|
def tree_depth(self):
|
|
"""Profondeur dans l'arbre (0 = racine)."""
|
|
depth = 0
|
|
node = self
|
|
while node.parent_id:
|
|
depth += 1
|
|
node = node.parent
|
|
if depth > 20: # securite anti-boucle
|
|
break
|
|
return depth
|
|
|
|
@property
|
|
def tree_path(self):
|
|
"""Chemin hierarchique complet : 'Fenetres college > Fenetre B01 > Fenetre gauche'."""
|
|
parts = [self.name]
|
|
node = self
|
|
while node.parent_id:
|
|
node = node.parent
|
|
parts.insert(0, node.name)
|
|
if len(parts) > 20:
|
|
break
|
|
return ' > '.join(parts)
|
|
|
|
def count_descendants(self):
|
|
"""Nombre total de descendants (recursif)."""
|
|
count = self.children.count()
|
|
for child in self.children:
|
|
count += child.count_descendants()
|
|
return count
|
|
|
|
def is_leaf(self):
|
|
"""True si c'est une feuille (pas d'enfants)."""
|
|
return self.children.count() == 0
|
|
|
|
@property
|
|
def needs_deep_work(self):
|
|
"""Nécessite une intervention profonde."""
|
|
return False
|
|
|
|
@property
|
|
def children_count(self):
|
|
"""Nombre d'enfants directs."""
|
|
return Equipment.query.filter_by(parent_id=self.id).count()
|
|
|
|
@property
|
|
def total_quantity(self):
|
|
"""Quantite physique declaree, individualisation comprise."""
|
|
return self.quantity
|
|
|
|
@property
|
|
def individualized_quantity(self):
|
|
"""Nombre d'unites du groupe disposant deja de leur propre fiche."""
|
|
if not self.is_group:
|
|
return 0
|
|
return sum(
|
|
max(child.quantity or 1, 1)
|
|
for child in self.children
|
|
if not child.is_group
|
|
)
|
|
|
|
@property
|
|
def unindividualized_quantity(self):
|
|
"""Nombre d'unites encore gerees uniquement sous forme quantitative."""
|
|
if not self.is_group:
|
|
return 0
|
|
return max((self.quantity or 0) - self.individualized_quantity, 0)
|
|
|
|
@property
|
|
def total_children(self):
|
|
"""Nombre total d'enfants."""
|
|
return self.children_count
|
|
|
|
@property
|
|
def is_low_stock(self):
|
|
"""Vérifie si le stock est faible."""
|
|
return False
|
|
|
|
@property
|
|
def is_individual(self):
|
|
"""Vérifie si c'est un élément individuel."""
|
|
return self.parent_id is not None
|
|
|
|
@property
|
|
def effective_room(self):
|
|
"""Salle propre, ou salle heritee du parent si elle n'est pas definie."""
|
|
if self.room_id and self.room:
|
|
return self.room
|
|
if self.is_individual and self.parent:
|
|
return self.parent.effective_room
|
|
return self.room
|
|
|
|
def create_scheduled_tasks_from_lot(self):
|
|
"""Crée les tâches planifiées à partir des tâches du lot héritées."""
|
|
from .maintenance import LotTask
|
|
from .planning import ScheduledTask
|
|
from datetime import date, timedelta
|
|
|
|
# Utiliser le lot effectif (hérité du parent si non défini)
|
|
effective_lot_id = self.effective_lot_id
|
|
|
|
if not effective_lot_id:
|
|
return []
|
|
|
|
created_tasks = []
|
|
|
|
# Récupérer les tâches du lot
|
|
lot_tasks = LotTask.query.filter_by(lot_id=effective_lot_id).all()
|
|
for lt in lot_tasks:
|
|
# Vérifier si une tâche planifiée existe déjà
|
|
existing = ScheduledTask.query.filter_by(
|
|
lot_task_id=lt.id,
|
|
equipment_id=self.id
|
|
).first()
|
|
|
|
if not existing:
|
|
# Calculer la première date planifiée
|
|
scheduled_date = date.today()
|
|
if lt.jours_entre_interventions:
|
|
scheduled_date = date.today() + timedelta(days=lt.jours_entre_interventions)
|
|
|
|
# Créer la tâche planifiée
|
|
scheduled_task = ScheduledTask(
|
|
lot_task_id=lt.id,
|
|
equipment_id=self.id,
|
|
room_id=self.room_id or (self.effective_room.id if self.effective_room else None),
|
|
scheduled_date=scheduled_date,
|
|
estimated_duration=lt.duree_minutes,
|
|
status='planned'
|
|
)
|
|
db.session.add(scheduled_task)
|
|
created_tasks.append(scheduled_task)
|
|
|
|
if created_tasks:
|
|
db.session.commit()
|
|
|
|
return created_tasks
|
|
|
|
def __repr__(self):
|
|
return f"<Equipment {self.name}>"
|
|
|
|
|
|
class EquipmentDocument(db.Model):
|
|
"""Document attaché à un équipement."""
|
|
__tablename__ = "equipment_documents"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=False)
|
|
filename = db.Column(db.String(255), nullable=False)
|
|
filepath = db.Column(db.String(500), nullable=False)
|
|
description = db.Column(db.Text, nullable=True)
|
|
uploaded_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
uploaded_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
|
|
|
|
# Relations
|
|
equipment = db.relationship("Equipment", back_populates="documents")
|
|
|
|
def __repr__(self):
|
|
return f"<EquipmentDocument {self.filename}>"
|
|
|
|
|
|
class EquipmentRoomHistory(db.Model):
|
|
"""Historique des déplacements d'équipements."""
|
|
__tablename__ = "equipment_room_history"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=False)
|
|
old_room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True)
|
|
new_room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True)
|
|
changed_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
changed_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
|
|
reason = db.Column(db.String(255), nullable=True)
|
|
|
|
# Relations
|
|
old_room = db.relationship("Room", foreign_keys=[old_room_id])
|
|
new_room = db.relationship("Room", foreign_keys=[new_room_id])
|
|
|
|
def __repr__(self):
|
|
return f"<EquipmentRoomHistory {self.equipment_id}>"
|
|
|
|
|
|
class RoomConstraint(db.Model):
|
|
"""Contrainte d'indisponibilite d'une salle (cours PRONOTE, horaire special, etc.)."""
|
|
__tablename__ = "room_constraints"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True) # NULL = toutes les salles
|
|
constraint_type = db.Column(db.String(20), nullable=False) # 'time', 'occupancy', 'event'
|
|
day_of_week = db.Column(db.Integer, nullable=True) # 0=Dimanche, 1=Lundi... 6=Samedi, NULL=tous les jours
|
|
start_time = db.Column(db.Time, nullable=True)
|
|
end_time = db.Column(db.Time, nullable=True)
|
|
reason = db.Column(db.String(255), nullable=True)
|
|
source = db.Column(db.String(50), default='manual') # 'pronote', 'manual', 'vacances'
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
effective_date_start = db.Column(db.Date, nullable=True)
|
|
effective_date_end = db.Column(db.Date, 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))
|
|
|
|
room = db.relationship("Room", backref="constraints_list")
|
|
|
|
@property
|
|
def day_name(self):
|
|
names = {0: 'Dimanche', 1: 'Lundi', 2: 'Mardi', 3: 'Mercredi', 4: 'Jeudi', 5: 'Vendredi', 6: 'Samedi'}
|
|
return names.get(self.day_of_week, 'Tous')
|
|
|
|
@property
|
|
def is_time_restriction(self):
|
|
return self.constraint_type == 'time'
|
|
|
|
@property
|
|
def is_occupancy(self):
|
|
return self.constraint_type == 'occupancy'
|
|
|
|
def applies_to_date(self, check_date):
|
|
"""Verifie si la contrainte s'applique a cette date."""
|
|
if not self.is_active:
|
|
return False
|
|
if self.day_of_week is not None and check_date.weekday() != self.day_of_week:
|
|
return False
|
|
if self.effective_date_start and check_date < self.effective_date_start:
|
|
return False
|
|
if self.effective_date_end and check_date > self.effective_date_end:
|
|
return False
|
|
return True
|
|
|
|
def conflicts_with(self, check_date, start_time, end_time):
|
|
"""Verifie si cette contrainte bloque un creneau donne."""
|
|
if not self.applies_to_date(check_date):
|
|
return False
|
|
if self.is_occupancy:
|
|
return True # Salle occupee toute la journee
|
|
if self.start_time and self.end_time:
|
|
# Chevauchement d'horaires
|
|
return start_time < self.end_time and end_time > self.start_time
|
|
return False
|
|
|
|
def __repr__(self):
|
|
return f"<RoomConstraint {self.room_id} {self.constraint_type}>"
|
|
|
|
|
|
def is_room_available(room_id, check_date, start_time, end_time):
|
|
"""Verifie si une salle est disponible a une date/heure donnee.
|
|
|
|
Croise les contraintes RoomConstraint.
|
|
Retourne (available: bool, reason: str or None).
|
|
"""
|
|
constraints = RoomConstraint.query.filter(
|
|
db.or_(RoomConstraint.room_id == room_id, RoomConstraint.room_id.is_(None)),
|
|
RoomConstraint.is_active == True
|
|
).all()
|
|
|
|
for c in constraints:
|
|
if c.conflicts_with(check_date, start_time, end_time):
|
|
room_name = c.room.name if c.room else 'Toutes salles'
|
|
return False, f"{room_name}: {c.reason or c.constraint_type}"
|
|
|
|
return True, None
|
|
|
|
|
|
class EquipmentRestriction(db.Model):
|
|
"""Restrictions d'accès pour un équipement."""
|
|
__tablename__ = "equipment_restrictions"
|
|
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=False)
|
|
|
|
# Type de restriction
|
|
restriction_type = db.Column(db.String(20), nullable=False) # 'time' ou 'room_occupancy'
|
|
|
|
# Pour restriction par horaire
|
|
start_time = db.Column(db.Time, nullable=True) # Heure de début
|
|
end_time = db.Column(db.Time, nullable=True) # Heure de fin
|
|
days_of_week = db.Column(db.String(20), nullable=True) # "0,1,2,3,4,5,6" pour les jours
|
|
|
|
# Pour restriction par occupation de salle
|
|
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True) # Salle à surveiller
|
|
|
|
# Message explicatif
|
|
message = db.Column(db.String(255), nullable=True) # "Intervention impossible pendant les cours"
|
|
|
|
# Métadonnées
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
|
created_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
|
|
|
|
# Relations
|
|
equipment = db.relationship("Equipment", backref="restrictions")
|
|
room = db.relationship("Room", backref="restrictions")
|
|
|
|
def days_list(self):
|
|
"""Retourne la liste des jours."""
|
|
if not self.days_of_week:
|
|
return []
|
|
days_names = {0: 'Dimanche', 1: 'Lundi', 2: 'Mardi', 3: 'Mercredi', 4: 'Jeudi', 5: 'Vendredi', 6: 'Samedi'}
|
|
return [days_names.get(int(d), '') for d in self.days_of_week.split(',') if d.isdigit()]
|
|
|
|
def is_time_restriction(self):
|
|
return self.restriction_type == 'time'
|
|
|
|
def is_room_occupancy_restriction(self):
|
|
return self.restriction_type == 'room_occupancy'
|
|
|
|
def __repr__(self):
|
|
return f"<EquipmentRestriction {self.equipment_id} - {self.restriction_type}>"
|