USER
si on supprime la 2eme instantacion dans self.thumbnailManager = ThumbnailManager(parent=self, max_workers=max_workers, cache_size=cache_size) dans class CustomFileSystemModelForList(QFileSystemModel):
les miniatures ne se génèrent pas dans la vue de liste , pourquoi ?
import logging, os, threading, time, sys, cProfile, pstats, io, re, math
from PyQt5 import QtWidgets
from PyQt5.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QPushButton, QTreeView, QFileSystemModel, QListView, QDockWidget, QComboBox, QLineEdit, QDateEdit, QGraphicsView,
QGraphicsScene, QGraphicsPixmapItem, QSplitter, QLabel, QFileDialog, QGraphicsPathItem, QGraphicsItemGroup, QGraphicsTextItem, QGraphicsLineItem, QGraphicsEllipseItem, QToolBar, QAction,QToolButton, QGraphicsPolygonItem,
QMenu,QMessageBox, QHeaderView, QGraphicsItem, QGraphicsRectItem, )
from PyQt5 import QtGui
from PyQt5.QtGui import QVector2D, QPixmap, QImageReader, QImage, QPainter, QIcon, QPainterPath, QTransform, QPolygonF, QPen, QBrush, QColor, QFont, QFontMetrics
from unittest.mock import MagicMock
from PyQt5.QtCore import (
QDir, Qt, QSize, QRectF, pyqtSignal, QLineF, QTimer, QPointF, QPoint, QModelIndex, QSortFilterProxyModel, QAbstractProxyModel,QItemSelectionModel, QRegularExpression, QAbstractListModel, QFileInfo,QObject,
QThread, QDate, QMetaObject, QSemaphore, QThreadPool, QRunnable, QFileSystemWatcher,Q_ARG, pyqtSlot, QEvent, QMutex, QMutexLocker,QRecursiveMutex, QVariant)
from PyQt5.QtChart import QChart, QSplineSeries, QChartView
from PIL import Image, ImageQt
import ezdxf
import ezdxf.query
from ezdxf.entities import DXFEntity
from math import sin, cos, radians
from uuid import uuid4
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from collections import OrderedDict, deque
import matplotlib.pyplot as plt
from enum import Enum
import whoosh
from whoosh.index import create_in, open_dir, exists_in, EmptyIndexError
from whoosh.fields import Schema, TEXT, ID, DATETIME, STORED
from whoosh.qparser import QueryParser
from whoosh.query import Term, Or, DateRange, And, Every, FuzzyTerm, Prefix
from whoosh.writing import AsyncWriter
from datetime import datetime
import win32api,win32con
import shutil
# Définition des constantes
MAX_RETRIES = 3 # Nombre maximum de tentatives pour traiter un fichier
INDEX_DIR = os.path.normpath(r"C:\index_dir") # Chemin d'accès centralisé à l'index Whoosh
if os.path.exists(INDEX_DIR):
print("Le chemin INDEX_DIR est valide.")
# Ajoutez votre logique pour surveiller le dossier ici
else:
print("Le chemin INDEX_DIR n'est pas valide.")
WATCHED_DIR = os.path.normpath(r"C:\DocSoufiane")# Constante pour le répertoire spécifique à surveiller
if os.path.exists(WATCHED_DIR):
print("Le chemin WATCHED_DIR est valide.")
# Ajoutez votre logique pour surveiller le dossier ici
else:
print("Le chemin WATCHED_DIR n'est pas valide.")
ALLOWED_EXTENSIONS = ['.dxf', '.png', '.jpg', '.jpeg', '.bmp']
# Setup logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
def handle_uncaught_exception(exc_type, exc_value, exc_traceback):
"""Global exception handler."""
logger.critical("Unhandled exception!", exc_info=(exc_type, exc_value, exc_traceback))
sys.__excepthook__(exc_type, exc_value, exc_traceback) # Call the default exception handler
# Install the global exception handler
sys.excepthook = handle_uncaught_exception
class CancelledError(Exception):
pass
class RequestState(Enum):
PENDING = 'pending'
IN_PROGRESS = 'in_progress'
PROCESSED = 'processed'
class RequestThread(QThread):
"""Thread dédié à la gestion de la file d'attente des requêtes de vignettes."""
thumbnailRequested = pyqtSignal(str) # Signal pour transmettre les requêtes au ProcessingThread
def __init__(self, parent=None):
super().__init__(parent)
self.request_queue = deque()
self.lock = QMutex()
def run(self):
"""Boucle principale du thread.
Attend les demandes de vignettes et les transmet au thread de traitement.
"""
while True:
file_path = None
with QMutexLocker(self.lock): # Verrouiller l'accès à la file d'attente
if self.request_queue:
file_path = self.request_queue.popleft()
if file_path:
logger.debug(f"RequestThread: Sending thumbnail request for {file_path}")
self.thumbnailRequested.emit(file_path) # Émettre le signal pour demander la vignette
else:
time.sleep(0.1) # Pause de 100 ms
@pyqtSlot(str)
def enqueue_request(self, file_path):
"""Ajoute une requête de vignette à la file d'attente.
Args:
file_path (str): Chemin du fichier pour lequel une vignette est demandée.
"""
with QMutexLocker(self.lock): # Verrouiller l'accès à la file d'attente
if file_path not in self.request_queue:
self.request_queue.append(file_path)
logger.debug(f"RequestThread: Request queued for {file_path}. Queue: {list(self.request_queue)}")
class ProcessingThread(QThread):
"""Thread dédié au traitement des requêtes de vignettes et à la gestion du cache."""
thumbnailReady = pyqtSignal(str, QPixmap, bool) # Signal pour notifier la disponibilité des vignettes
requestProcessed = pyqtSignal(str, bool) # Signal pour signaler le traitement d'une requête
def __init__(self, thumbnail_manager, max_workers=10, parent=None):
super().__init__(parent)
self.thumbnail_manager = thumbnail_manager # Référence au ThumbnailManager
self.max_workers = max_workers
self.request_queue = deque()
self.lock = QMutex()
self.pending_requests_lock = QMutex() # Verrou spécifique pour pending_requests
self.semaphore = QSemaphore(max_workers) # Sémaphore pour limiter le nombre de workers
self.thumbnail_manager.requestThumbnail.connect(self.handle_thumbnail_request)
self.pending_requests = set()
def run(self):
"""Boucle principale du thread.
Traite les requêtes de vignettes en lançant des ThumbnailWorker.
"""
while True:
file_path = None
with QMutexLocker(self.lock): # Verrouiller l'accès à la file d'attente
if self.request_queue:
file_path = self.request_queue.popleft()
if file_path:
with QMutexLocker(self.pending_requests_lock): # Protéger l'accès à pending_requests
if file_path not in self.pending_requests:
self.pending_requests.add(file_path)
if self.semaphore.tryAcquire(): # Acquérir un sémaphore si disponible
logger.debug(f"ProcessingThread: Processing request for {file_path}")
worker = ThumbnailWorker(file_path, self.cache) # Passer le cache ici
worker.signals.thumbnailReady.connect(self.thumbnailReady.emit) # Connecter le signal du worker
worker.signals.finished.connect(self.worker_finished)
worker.signals.errorOccurred.connect(self.handle_worker_error)
self.thumbnail_manager.threadpool.start(worker)
else:
# Si aucun sémaphore n'est disponible, remettre la requête en file d'attente
with QMutexLocker(self.lock):
self.request_queue.appendleft(file_path)
logger.debug(f"ProcessingThread: No semaphore available, request for {file_path} requeued")
else:
# Pause pour éviter la surcharge CPU lorsque la file d'attente est vide
time.sleep(0.1)
@pyqtSlot(str)
def handle_thumbnail_request(self, file_path):
"""Ajoute une requête de vignette à la file d'attente.
Args:
file_path (str): Chemin du fichier pour lequel une vignette est demandée.
"""
with QMutexLocker(self.lock): # Verrouiller l'accès à la file d'attente
# Vérifier si le fichier est déjà dans la file d'attente ou en cours de traitement
with QMutexLocker(self.pending_requests_lock): # Utiliser un verrou pour pending_requests
if file_path not in self.request_queue and file_path not in self.pending_requests:
self.request_queue.append(file_path)
logger.debug(f"ProcessingThread: Request queued for {file_path}. Queue: {list(self.request_queue)}")
@pyqtSlot()
def worker_finished(self, file_path):
"""Appelée lorsque le travailleur a terminé, pour libérer le sémaphore et mettre à jour les requêtes en attente."""
with QMutexLocker(self.pending_requests_lock):
self.pending_requests.discard(file_path)
self.semaphore.release()
@pyqtSlot(str, Exception)
def handle_worker_error(self, file_path, error):
"""Gérer les erreurs survenues dans le travailleur."""
logger.error(f"Error processing {file_path}: {error}")
# Retirer le fichier des requêtes en attente même en cas d'erreur
with QMutexLocker(self.pending_requests_lock):
self.pending_requests.discard(file_path)
self.semaphore.release()
class ThumbnailManager(QObject):
"""Gère la coordination des vignettes, le cache et la communication entre les threads."""
thumbnailReady = pyqtSignal(str, QPixmap, bool) # Chemin du fichier, vignette (ou None), succès
requestThumbnail = pyqtSignal(str)
def __init__(self, parent=None, max_workers=10, cache_size=5000):
super().__init__(parent)
self.cache = ThumbnailCache(cache_size)
self.request_thread = RequestThread()
self.processing_thread = ProcessingThread(self)
self.request_thread.thumbnailRequested.connect(self.processing_thread.handle_thumbnail_request)
self.request_thread.start()
self.processing_thread.start()
self.threadpool = QThreadPool()
self.threadpool.setMaxThreadCount(max_workers)
self.lock = QMutex() # Verrou pour protéger l'accès au cache et aux workers
self.workers = {} # Dictionnaire pour suivre les workers actifs
@pyqtSlot(str, QPixmap, bool)
def handle_thumbnail_ready(self, file_path, thumbnail, success):
"""Gère la réception des vignettes générées par les workers."""
with QMutexLocker(self.lock):
if success:
#logger.debug(f"handle_thumbnail_ready: Mise en cache de la vignette pour {file_path}")
self.cache[file_path] = thumbnail # Mettre en cache la vignette
if file_path in self.workers:
del self.workers[file_path] # Supprimer le worker terminé
#logger.debug(f"handle_thumbnail_ready: Worker supprimé pour {file_path}")
#logger.debug(f"handle_thumbnail_ready: Émission du signal thumbnailReady pour {file_path}")
self.thumbnailReady.emit(file_path, thumbnail, success) # Émettre le signal pour mettre à jour la vue
def start_worker(self, file_path):
"""Lance un worker pour générer une vignette pour le fichier donné."""
with QMutexLocker(self.lock):
if file_path in self.cache: # Vérifier si la vignette est déjà dans le cache
#logger.debug(f"ThumbnailManager: Thumbnail found in cache for {file_path}")
self.thumbnailReady.emit(file_path, self.cache[file_path], True)
return
if file_path in self.workers: # Vérifier si un worker est déjà en cours pour ce fichier
logger.debug(f"ThumbnailManager: Thumbnail generation already in progress for {file_path}")
return
#logger.debug(f"ThumbnailManager: Starting thumbnail worker for {file_path}")
worker = ThumbnailWorker(file_path, self.cache)
worker.signals.thumbnailReady.connect(self.handle_thumbnail_ready)
worker.signals.finished.connect(self.worker_finished)
self.workers[file_path] = worker # Ajouter le worker au dictionnaire
self.threadpool.start(worker)
@pyqtSlot()
def worker_finished(self):
"""Slot appelé lorsqu'un worker a terminé."""
worker = self.sender()
if worker is not None and hasattr(worker, 'file_path'): # Check if worker is valid and has file_path
file_path = worker.file_path
logger.debug(f"ThumbnailManager: Worker finished for {file_path}")
with QMutexLocker(self.lock):
if file_path in self.workers:
del self.workers[file_path]
logger.debug(f"worker_finished: Worker supprimé pour {file_path}")
else:
logger.warning("ThumbnailManager: Worker finished but sender is None or doesn't have file_path")
class ThumbnailCache(OrderedDict):
"""Cache pour les vignettes générées."""
def __init__(self, cache_size=5000):
super().__init__()
self.cache_size = cache_size
logger.debug(f"ThumbnailCache: Initialisé avec une taille maximale de {cache_size} vignettes.")
def __getitem__(self, key):
logger.debug(f"ThumbnailCache: __getitem__ appelé pour {key}")
try:
value = super().__getitem__(key)
self.move_to_end(key)
logger.debug(f"Thumbnail retrieved from cache for: {key}")
return value
except KeyError:
logger.warning(f"ThumbnailCache: Vignette non trouvée dans le cache pour {key}")
raise # Relève l'exception pour que le code appelant puisse la gérer
def get(self, key):
#logger.debug(f"ThumbnailCache: get appelé pour {key}")
value = super().pop(key, None)
if value is not None:
self[key] = value # Remettre à la fin pour indiquer un accès récent
#logger.debug(f"(ThumbnailCache get): Thumbnail found in cache for: {key}")
#else:
#logger.debug(f"(ThumbnailCache get): Thumbnail not found in cache for: {key}")
return value
def __setitem__(self, key, value):
#logger.debug(f"ThumbnailCache: __setitem__ appelé pour {key}")
super().__setitem__(key, value)
self.move_to_end(key) # Déplacer l'élément à la fin (le plus récemment utilisé)
if len(self) > self.cache_size:
oldest = next(iter(self))
logger.debug(f"ThumbnailCache: Suppression de l'élément le plus ancien {oldest}")
self.pop(oldest) # Supprimer l'élément le moins récemment utilisé
class CustomFileSystemModelForList(QFileSystemModel):
"""Modèle de système de fichiers personnalisé pour afficher les vignettes dans une vue de liste."""
searchFinished = pyqtSignal(list) # results: List[str]
rootPathChanged = pyqtSignal(str) # new_root_path: str
#dataChanged = pyqtSignal(QModelIndex, QModelIndex) # index: QModelIndex
modelUpdated = pyqtSignal(QModelIndex) # index: QModelIndex # Signal to notify that the model has been updated
directoryLoaded = pyqtSignal(str) # path: str
#thumbnailUpdated = pyqtSignal(str, QPixmap, bool) # file_path: str, thumbnail: QPixmap, success: bool
thumbnailRequested = pyqtSignal(str) #
def __init__(self, parent=None, indexer=None, max_workers=10, cache_size=5000):
logger.debug("Beginning of __init__ function in CustomFileSystemModelForList")
super().__init__(parent)
self.allowed_extensions = ALLOWED_EXTENSIONS
self.setRootPath(os.path.normpath(WATCHED_DIR))
# Connexion des signaux pour le débogage
#self.searchFinished.connect(lambda results: print(f"Search finished signal received with results: {results}"))
#self.rootPathChanged.connect(lambda path: print(f"Root path changed signal received with path: {path}"))
#self.dataChanged.connect(lambda top_left, bottom_right: print(f"Data changed signal received for indexes: {top_left}, {bottom_right}"))
#self.rootPathChanged.connect(lambda path: logger.debug(f"Root path changed: {path}"))
#self.modelUpdated.connect(lambda index: print(f"Model updated signal received for index: {index}"))
self.thumbnailManager = ThumbnailManager(parent=self, max_workers=max_workers, cache_size=cache_size)
# Connexion du signal thumbnailReady au slot correspondant de ThumbnailManager
self.thumbnailManager.thumbnailReady.connect(self.on_thumbnail_ready)
# Connexion explicite du signal thumbnailRequested
self.thumbnailRequested.connect(self.thumbnailManager.start_worker)
# Connexion du signal requestProcessed au slot correspondant de FileViewer
#self.processing_thread.requestProcessed.connect(self.handle_request_processed)
self.parent = parent
self.file_list = QListView()
self.file_list.setStyleSheet("QListView::item { margin: 5px; }")
self.criteria = ""
self._search_timer = QTimer(self)
self._search_timer.setSingleShot(True)
self._last_criteria = ""
self.indexer = indexer
if self.indexer:
# Connect the indexUpdated signal with an anonymous lambda function to provide rootPath
self.indexer.indexUpdated.connect(lambda: QTimer.singleShot(1000, # Délai de 1 seconde
self.add_files_to_watcher))
# Cache resources
script_directory = os.path.dirname(os.path.abspath(__file__))
icon_path = os.path.join(script_directory, "dossier.png")
self.folder_icon = QIcon(QPixmap(icon_path))
if self.folder_icon.isNull():
logger.error(f"Failed to load folder icon from {icon_path}")
self.thumbnailManager = ThumbnailManager(parent=self, max_workers=max_workers, cache_size=cache_size) # Créer une instance de ThumbnailManager
self.thumbnailManager.thumbnailReady.connect(self.on_thumbnail_ready)
logger.debug("Successfully connected thumbnailReady signal to on_thumbnail_ready slot.")
self.file_watcher = QFileSystemWatcher(self) # Initialize file_watcher
self.directoryLoaded.connect(self.on_directory_loaded)
#self.file_watcher.fileChanged.connect(self.handle_file_change) # Connect to the signal
logger.debug("End of __init__ function in CustomFileSystemModelForList")
def event(self, event):
return super().event(event) # Revenir à la méthode par défaut
def rowCount(self, parent=QModelIndex()):
#logger.debug(f"rowCount called with parent: {parent}")
return super().rowCount(parent)
def index(self, *args, **kwargs):
#logger.debug(f"index called with row: {row}, column: {column}, parent: {parent}")
try:
#result = super().index(row, column, parent)
#logger.debug(f"CustomFileSystemModelForList.index: Returning index: {result}")
if len(args) == 1 and isinstance(args[0], str):
# Appel avec path
path = args[0]
column = kwargs.get('column', 0)
#logger.debug(f"index called with path: {path}, column: {column}")
return super().index(path, column)
#return super().index(row, column, parent)
elif len(args) == 2 and isinstance(args[0], str) and isinstance(args[1], int):
# Appel avec path et column
path = args[0]
column = args[1]
#logger.debug(f"index called with path: {path}, column: {column}")
return super().index(path, column)
else:
# Appel avec row, column, parent
return super().index(*args, **kwargs)
except Exception as e:
logger.error(f"CustomFileSystemModelForList.index: An error occurred: {e}")
raise # Remonte l'exception pour une gestion plus haut dans le code si nécessaire
def indexFromPath(self, path):
"""Helper function to get the index from a path."""
#logger.debug(f"Entering CustomFileSystemModelForList.indexFromPath with path: {path}")
result = super().index(path)
#logger.debug(f"CustomFileSystemModelForList.indexFromPath: Returning index: {result}")
return result
def add_files_to_watcher(self):
"""
Ajoute les fichiers du répertoire surveillé (WATCHED_DIR) et de ses sous-répertoires
au `QFileSystemWatcher` pour surveiller les modifications.
"""
logger.debug(f"Entering CustomFileSystemModelForList.add_files_to_watcher")
try:
if not self.file_watcher:
logger.warning("file_watcher is not initialized.")
return
# Assurez-vous que le répertoire surveillé est correctement normalisé
normalized_watched_dir = os.path.normpath(WATCHED_DIR)
# Utiliser QDir pour lister les fichiers dans le répertoire surveillé
dir = QDir(normalized_watched_dir)
dir.setNameFilters([f"*{ext}" for ext in self.allowed_extensions])
dir.setFilter(QDir.Files | QDir.NoSymLinks | QDir.NoDotAndDotDot)
# Obtenir tous les fichiers correspondants dans le répertoire et ses sous-répertoires
file_info_list = dir.entryInfoList(QDir.AllEntries | QDir.NoDotAndDotDot, QDir.Name)
allowed_file_paths = []
for file_info in file_info_list:
file_path_str = file_info.absoluteFilePath()
#logger.debug(f"Checking file path: {file_path_str}")
# Vérifiez si le fichier est caché ou système
if file_info.isHidden() or (win32api.GetFileAttributes(file_path_str) & win32con.FILE_ATTRIBUTE_SYSTEM):
logger.debug(f" - Skipped: Is hidden or system file")
continue
# Vérifier l'extension du fichier
extension = os.path.splitext(file_path_str)[1].lower()
if extension not in self.allowed_extensions:
logger.debug(f" - Skipped: Extension {extension} not allowed")
continue
allowed_file_paths.append(file_path_str)
if not allowed_file_paths:
logger.warning("No valid file paths found to add to QFileSystemWatcher.")
logger.warning(f" - Root path: {WATCHED_DIR}")
logger.warning(f" - Allowed extensions: {self.allowed_extensions}")
else:
self.file_watcher.addPaths(allowed_file_paths)
logger.info(f"{len(allowed_file_paths)} files added to QFileSystemWatcher.")
except Exception as e:
logger.critical(f"Unhandled exception in add_files_to_watcher: {e}", exc_info=True)
logger.debug("Exiting CustomFileSystemModelForList.add_files_to_watcher")
def on_directory_loaded(self, path):
"""
Gère les actions à effectuer après le chargement d'un répertoire.
"""
logger.info(f"on_directory_loaded / Répertoire chargé (CustomFileSystemModelForList) : {path}")
if self.indexer:
normalized_path = os.path.normpath(path)
if normalized_path.startswith(os.path.normpath(WATCHED_DIR)):
self.indexer.buildIndex()
logger.debug("add_files_to_watcher.on_directory_loaded(self, path) / self.indexer.buildIndex() called")
#self.add_files_to_watcher(path)
else:
logger.warning("CustomFileSystemModelForList/ on_directory_loaded; Indexer non disponible dans CustomFileSystemModelForList.")
def data(self, index, role=Qt.DisplayRole):
#logger.debug(f"CustomFileSystemModelForList.data called for index {index} with role {role}")
if role in (Qt.SizeHintRole, Qt.FontRole):
return None
try:
if not index.isValid():
return None
if role not in [Qt.DisplayRole, Qt.DecorationRole, Qt.ToolTipRole]:
return None
return self.standard_data_method(index, role)
except Exception as e:
logger.exception(f"CustomFileSystemModelForList.data: Exception occurred: {e}")
return None
finally:
pass
#logger.debug(f"CustomFileSystemModelForList.data finished for index {index} with role {role}")
def standard_data_method(self, index, role):
"""Méthode standard pour fournir les données du modèle."""
if role in (Qt.SizeHintRole, Qt.FontRole): # Optimisation : ignorer ces rôles
return None
file_path = self.filePath(index)
#logger.debug(f"CustomFileSystemModelForList.standard_data_method called for {file_path}, role: {role}")
#logger.debug(f"standard_data_method: Current thread: {QThread.currentThread()}")
try:
if not index.isValid():
return None
#logger.debug(f"Entering role check for {file_path}")
if role not in (Qt.DisplayRole, Qt.DecorationRole, Qt.ToolTipRole):
#logger.debug(f"Unsupported role {role} for {file_path}, returning None")
return None
#logger.debug(f"Processing role {role} for {file_path}")
if role == Qt.DisplayRole:
filename = os.path.basename(file_path)
#logger.debug(f"DisplayRole: Original filename {filename}")
if file_path in QDir.drives():
filename = file_path
#logger.debug(f"DisplayRole: Filename is a drive {filename}")
longueur_max_titre = 25
if len(filename) > longueur_max_titre:
filename = filename[:longueur_max_titre - 3] + "..."
return filename
elif role == Qt.ToolTipRole:
#logger.debug(f"ToolTipRole: Returning file path {file_path}")
return file_path
elif role == Qt.DecorationRole and index.column() == 0:
#logger.debug(f"CustomFileSystemModelForList: Standard data method called ; Requesting thumbnail for {file_path} , role: {role}")
if os.path.isdir(file_path):
#logger.debug(f"DecorationRole: {file_path} is a directory, returning folder icon")
return self.folder_icon
elif file_path.lower().endswith(('.dxf', '.png', '.jpg', '.jpeg')):
thumbnail = self.thumbnailManager.cache.get(file_path) # Vérifier le cache
if thumbnail is None:
with QMutexLocker(self.thumbnailManager.processing_thread.pending_requests_lock): # Utiliser le verrou spécifique
# Vérifier les requêtes en attente dans ThumbnailManager
if file_path not in self.thumbnailManager.processing_thread.pending_requests:
self.thumbnailManager.processing_thread.pending_requests.add(file_path)
if QThread.currentThread() == QApplication.instance().thread():
self.thumbnailRequested.emit(file_path)
else:
# Émettre le signal depuis le thread principal en utilisant un timer
QTimer.singleShot(0, lambda: self.thumbnailRequested.emit(file_path))
return QPixmap() # Retourner une image vide en attendant
else:
return thumbnail.scaled(160, 120, Qt.KeepAspectRatio, Qt.SmoothTransformation) # Redimensionner la vignette
else:
return None # Aucun icône pour les autres types de fichiers
except Exception as e:
logger.exception(f"CustomFileSystemModelForList.data: Exception occurred: {e}")
return None
@pyqtSlot(str, QPixmap, bool)
def on_thumbnail_ready(self, file_path, thumbnail, success):
"""Slot appelé lorsque la vignette est prête."""
#logger.debug(f"CustomFileSystemModelForList: on_thumbnail_ready() - file_path: {file_path}, success: {success}")
if success:
self.update_thumbnail_cache(file_path, thumbnail)
else:
logger.warning(f"on_thumbnail_ready: Échec de la génération de la vignette pour {file_path}")
@pyqtSlot(str, bool)
def handle_request_processed(self, file_path, success):
"""Gère le signal requestProcessed du ProcessingThread."""
if not success:
# Gérez l'échec de la génération de la vignette (réessayez, affichez un message d'erreur, etc.)
logger.error(f"CustomFileSystemModelForList: Failed to generate thumbnail for {file_path}")
else:
logger.debug(f"CustomFileSystemModelForList: Thumbnail successfully generated for {file_path}")
@pyqtSlot(str, QPixmap)
def update_thumbnail_cache(self, file_path, thumbnail):
"""Met à jour le cache des vignettes et émet le signal dataChanged dans le thread principal."""
#logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Entrée avec file_path: {file_path}")
# Utiliser QMutexLocker pour assurer le déverrouillage automatique
with QMutexLocker(self.thumbnailManager.lock):
#logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Verrou acquis")
if file_path not in self.thumbnailManager.cache or self.thumbnailManager.cache[file_path] != thumbnail:
#logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Vignette absente du cache ou différente, mise à jour en cours...")
self.thumbnailManager.cache[file_path] = thumbnail
#logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Vignette mise à jour dans le cache pour {file_path}")
else:
logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Vignette déjà présente dans le cache, aucune mise à jour nécessaire.")
# Émettre le signal dataChanged pour indiquer que les données ont changé
index = self.indexFromPath(file_path)
if index.isValid():
#logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Émission du signal dataChanged pour {file_path}")
self.dataChanged.emit(index, index, [Qt.DecorationRole]) # Émettre le signal dataChanged pour l'élément spécifique
#logger.debug(f"CustomFileSystemModelForList: update_thumbnail_cache() - Signal dataChanged émis pour {file_path}")
else:
logger.warning(f"CustomFileSystemModelForList: update_thumbnail_cache() - Index invalide pour {file_path}")
class CustomFileSystemModelForTree(QFileSystemModel):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Cache the folder icon for performance
script_directory = os.path.dirname(os.path.abspath(__file__))
icon_path = os.path.join(script_directory, "dossier.png")
self.folder_icon = QIcon(QPixmap(icon_path))
if self.folder_icon.isNull():
logger.error(f"Failed to load folder icon from {icon_path}")
def data(self, index, role):
if role == Qt.DecorationRole and index.column() == 0:
# Get the full path of the file or directory at the index
file_path = self.filePath(index)
# Check if it's a directory
if os.path.isdir(file_path):
return self.folder_icon
elif role == Qt.DisplayRole:
# Get the default display text
original_text = super().data(index, role)
return original_text
return None
# Background search worker class
class SearchWorker(QObject):
searchFinished = pyqtSignal(bool, list) # Signal to indicate search success
def __init__(self, indexer, index_dir):
super().__init__()
self.running = False
self.running_mutex = QMutex() # Mutex to protect self.running
self.rootPath = ""
self.criteria = ""
self.criteria_type = "filename"
self.start_date = None
self.end_date = None
self.allowed_extensions = ALLOWED_EXTENSIONS
self.extension_pattern = re.compile(r".*(" + "|".join([re.escape(ext) + "$" for ext in self.allowed_extensions]) + ")", re.IGNORECASE)
self.indexer = indexer
self.index_dir = os.path.normpath(index_dir)
logger.debug(f"SearchWorker initialized with index_dir: {self.index_dir}")
def stop(self):
"""Stop the ongoing search gracefully."""
self.running_mutex.lock()
running = self.running
self.running_mutex.unlock()
if not running:
logger.info("Search interrupted by user.")
return # Exit if the search was not running
logger.info("Search stopped by user.")
self.running = False # This will cause the search loop to exit
def setCriteria(self, criteria, criteria_type="filename"):
"""Sets the search criteria and its type."""
self.criteria = criteria.lower() # Convert criteria to lowercase
self.criteria_type = criteria_type
logger.debug(f"Search criteria set: {criteria} (type: {criteria_type})")
def setBasePath(self, path):
self.rootPath = os.path.normpath(path)
logger.debug(f"Base path for search set: {self.rootPath}")
def setDateRange(self, start_date, end_date):
self.start_date = start_date.toPyDate()
self.end_date = end_date.toPyDate()
logger.debug(f"Date range set: {self.start_date} - {self.end_date}")
# Pre-calculate timestamps for date comparison
self.start_date_timestamp = datetime.combine(self.start_date, datetime.min.time()).timestamp()
self.end_date_timestamp = datetime.combine(self.end_date, datetime.max.time()).timestamp()
def performSearch(self, retry=False):
logger.info("Starting search.")
logger.debug(f"Root path for search: {self.rootPath}")
# Verify root path
if not self.rootPath or not os.path.exists(self.rootPath):
logger.error("Base path not set or does not exist.")
self.searchFinished.emit(False, [])
return
# Validate dates
if self.start_date and self.end_date and self.start_date > self.end_date:
logger.error("Start date must be before end date.")
self.searchFinished.emit(False, [])
return
self.running_mutex.lock() # Lock before modifying
self.running = True
self.running_mutex.unlock() # Unlock after modification
results = []
criteria_lower = self.criteria.lower().strip()
if not criteria_lower:
logger.error("Search criteria is empty after processing. Aborting search.")
self.searchFinished.emit(False, [])
return
try:
# Check if within watched directory
normalized_root_path = os.path.normcase(os.path.normpath(self.rootPath))
normalized_watched_dir = os.path.normcase(os.path.normpath(WATCHED_DIR))
if os.path.commonpath([normalized_root_path, normalized_watched_dir]) == normalized_watched_dir:
logger.info("Using index-based search.")
self.index_based_search(results, criteria_lower, retry)
else:
logger.info("Using file system search.")
self.scan_directory(normalized_root_path, results, criteria_lower)
#logger.debug(f"Final search results: {results}")
except Exception as e:
logger.error(f"Error during search: {e}")
self.searchFinished.emit(False, [])
return
finally:
logger.info("Finalizing search.")
self.running_mutex.lock()
self.running = False
self.running_mutex.unlock()
self.searchFinished.emit(True, results)
logger.info(f"Search completed with {len(results)} results.")
def index_based_search(self, results, criteria_lower, retry):
try:
logger.info("Performing index-based search.")
ix = open_dir(self.index_dir)
with ix.searcher() as searcher:
query_parts = []
# Build query based on criteria type
if self.criteria_type == "filename":
if not criteria_lower.strip():
raise ValueError("Search criteria for filename is empty.")
query_part = Prefix("name", criteria_lower)
query_parts.append(query_part)
elif self.criteria_type in ("creation_date", "modification_date"):
date_field = "creation_time" if self.criteria_type == "creation_date" else "modification_time"
query_parts.append(DateRange(date_field, self.start_date, self.end_date))
# Filter by extensions using regex
if self.allowed_extensions: # Check if extensions are specified
query_parts.append(self.extension_pattern)
# Filter by path
query_parts.append(Prefix("path", os.path.normcase(os.path.normpath(self.rootPath))))
query = And(query_parts)
# Execute the search
hits = searcher.search(query, limit=None)
for hit in hits:
self.running_mutex.lock() # Protect access to self.running
running = self.running
self.running_mutex.unlock()
if not running:
break
results.append(hit["path"])
except EmptyIndexError:
if not retry:
logger.warning("Index directory is empty, creating index...")
self.indexer.buildIndex()
return self.performSearch(retry=True) # Retry after rebuilding index
else:
logger.error("Index is still empty after rebuilding.")
self.searchFinished.emit(False, [])
except Exception as e:
logger.error(f"Error during Whoosh search: {e}")
self.searchFinished.emit(False, [])
def scan_directory(self, directory, results, criteria_lower):
"""Recursively scan the directory for files matching the criteria."""
for entry in os.scandir(directory):
if entry.is_dir():
self.scan_directory(entry.path, results, criteria_lower) # Recursive call for subdirectories
elif entry.is_file():
self.process_file(entry.path, entry.name, results, criteria_lower)
def process_file(self, fullPath, file, results, criteria_lower):
try:
file_lower = file.lower()
_, extension = os.path.splitext(fullPath)
# Check extension using regex
if not self.extension_pattern.match(fullPath):
return
# Check criteria
if self.criteria_type == "filename" and criteria_lower in file_lower:
results.append(fullPath)
elif self.criteria_type in ("creation_date", "modification_date"):
date_to_check = os.path.getctime(fullPath) if self.criteria_type == "creation_date" else os.path.getmtime(fullPath)
if self.compare_dates(date_to_check) :
if criteria_lower in file_lower:
results.append(fullPath)
except Exception as e:
logger.error(f"Error processing file {fullPath}: {e}")
def compare_dates(self, file_timestamp):
return self.start_date_timestamp <= file_timestamp <= self.end_date_timestamp
class TreeProxyModel(QSortFilterProxyModel):
def filterAcceptsRow(self, source_row, source_parent):
index = self.sourceModel().index(source_row, 0, source_parent)
if self.sourceModel().isDir(index):
#logger.debug(f"Folder accepted for display: {index.data()}")
return True # Accept the entry as it is a folder
#logger.debug(f"Entry rejected (not a folder): {index.data()}")
return False # Reject the entry as it is not a folder
class FileListProxyModel(QSortFilterProxyModel):
def __init__(self, parent=None):
super(FileListProxyModel, self).__init__(parent)
self.filter_text = ""
self.filters = ['.dxf', '.png', '.jpg', '.jpeg']
self.recursiveSearchEnabled = True # Enabled by default
self.searchPaths = [] # Paths to use for filtering results
def setFilterText(self, text):
"""Set the filter text and invalidate the current filter."""
self.filter_text = text.lower()
self.invalidateFilter()
logger.debug(f"Filter text set: {self.filter_text}")
def setSearchPaths(self, paths):
"""Set the search paths for filtering."""
self.searchPaths = paths
self.invalidateFilter()
logger.debug(f"Search paths set: {self.searchPaths}")
def setRecursiveSearchEnabled(self, enabled):
"""Enable or disable recursive search."""
self.recursiveSearchEnabled = enabled
self.invalidateFilter()
logger.debug(f"Recursive search {'enabled' if enabled else 'disabled'}.")
def filterAcceptsRow(self, sourceRow, sourceParent):
sourceModel = self.sourceModel()
if not sourceModel:
return False
sourceIndex = sourceModel.index(sourceRow, 0, sourceParent)
if not sourceIndex.isValid():
return False
# Standard filtering logic
filepath = sourceModel.filePath(sourceIndex).lower()
isDirectory = sourceModel.isDir(sourceIndex)
isAcceptableFile = isDirectory or any(filepath.endswith(ext) for ext in self.filters)
filename = sourceModel.fileName(sourceIndex).lower()
if self.searchPaths:
if not any(filepath.startswith(path.lower()) for path in self.searchPaths):
#logger.debug(f"Path {filepath} rejected by search paths.")
return False
if self.filter_text:
filename = self.sourceModel().data(sourceIndex, Qt.DisplayRole).lower()
textMatch = self.filter_text in filename or self.filter_text in filepath
else:
textMatch = True
# Final decision for row acceptance
if isAcceptableFile and textMatch:
return True
# If recursive search is enabled and the index is a directory, check subdirectories
if self.recursiveSearchEnabled and isDirectory:
rowCount = sourceModel.rowCount(sourceIndex)
logger.debug(f"Recursive search enabled, number of subdirectories for {filepath}: {rowCount}")
for childRow in range(rowCount):
if self.filterAcceptsRow(childRow, sourceIndex):
logger.debug(f"Recursive acceptance for {filepath} due to subdirectory at row {childRow}")
return True
#logger.debug(f"Rejected {filepath}")
return False
class SearchResultsProxyModel(QAbstractProxyModel):
def __init__(self, parent=None):
super().__init__(parent)
self._search_results = [] # Stocke les chemins des fichiers trouvés
self._path_to_row = {} # Mapping from path to row index
self._cached_source_indexes = {} # Cache pour les index source
def setSearchResults(self, results):
"""Met à jour les résultats de la recherche et signale les changements au modèle."""
logger.debug(f"setSearchResults called with {len(results)} results.")
self.beginResetModel()
self._search_results = [os.path.normcase(r) for r in results]
self._path_to_row = {path: idx for idx, path in enumerate(self._search_results)}
self._cached_source_indexes.clear() # Vider le cache car les résultats ont changé
self.endResetModel()
logger.debug("setSearchResults: Model reset completed.")
def rowCount(self, parent=QModelIndex()):
"""Retourne le nombre de résultats de recherche."""
if parent.isValid(): # Si l'index parent est valide, on est dans un noeud enfant, donc 0 enfant
return 0
return len(self._search_results)
def filePath(self, index):
"""Retourne le chemin du fichier pour l'index donné, en déléguant au modèle source."""
source_index = self.mapToSource(index)
return self.sourceModel().filePath(source_index)
def columnCount(self, parent=QModelIndex()):
"""Retourne le nombre de colonnes du modèle source."""
return self.sourceModel().columnCount(parent) if self.sourceModel() else 0
def index(self, row, column, parent=QModelIndex()):
"""Retourne l'index du proxy pour la ligne, la colonne et le parent donnés."""
if not parent.isValid() and 0 <= row < len(self._search_results):
source_index = self.sourceModel().indexFromPath(self._search_results[row])
if source_index.isValid():
return self.createIndex(row, column, source_index)
else:
logger.error(f"Invalid source index for path: {self._search_results[row]}")
return QModelIndex()
def parent(self, index):
"""Retourne l'index parent, toujours invalide pour ce modèle."""
return QModelIndex()
def mapFromSource(self, sourceIndex):
"""Retourne l'index du proxy correspondant à l'index source."""
if sourceIndex.isValid():
source_file_path = os.path.normcase(os.path.normpath(self.sourceModel().filePath(sourceIndex)))
row = self._path_to_row.get(source_file_path)
if row is not None:
return self.createIndex(row, sourceIndex.column())
else:
return QModelIndex()
return QModelIndex()
def mapToSource(self, proxyIndex):
"""Retourne l'index source correspondant à l'index du proxy."""
if proxyIndex.isValid() and 0 <= proxyIndex.row() < len(self._search_results):
source_path = self._search_results[proxyIndex.row()]
if source_path in self._cached_source_indexes:
return self._cached_source_indexes[source_path]
source_index = self.sourceModel().indexFromPath(source_path)
if source_index.isValid():
self._cached_source_indexes[source_path] = source_index
return source_index
return QModelIndex()
def data(self, index, role=Qt.DisplayRole):
"""Retourne les données pour l'index donné et le rôle."""
if not index.isValid():
logger.debug("Invalid index in data method.")
return None
if not self.sourceModel():
logger.error("Source model is not set.")
return None
source_index = self.mapToSource(index) # Obtenir un nouvel index source valide
if not source_index.isValid():
logger.debug("Invalid source index in data method.")
return None
return self.sourceModel().data(source_index, role)
def setSourceModel(self, source_model):
logger.debug("SearchResultsProxyModel: setSourceModel called")
super().setSourceModel(source_model)
if source_model is not None:
source_model.dataChanged.connect(self.sourceDataChanged)
def sourceDataChanged(self, sourceTopLeft, sourceBottomRight, roles):
#logger.debug(f"SearchResultsProxyModel: sourceDataChanged called with sourceTopLeft={sourceTopLeft}, sourceBottomRight={sourceBottomRight}, roles={roles}")
# Mapper les index source vers les index du proxy
proxyTopLeft = self.mapFromSource(sourceTopLeft)
proxyBottomRight = self.mapFromSource(sourceBottomRight)
#logger.debug(f"SearchResultsProxyModel: Mapped proxyTopLeft={proxyTopLeft}, proxyBottomRight={proxyBottomRight}")
if proxyTopLeft.isValid() and proxyBottomRight.isValid():
#logger.debug("SearchResultsProxyModel: Emitting dataChanged from proxy model")
# Émettre le signal dataChanged du proxy avec les rôles appropriés
self.dataChanged.emit(proxyTopLeft, proxyBottomRight, roles)
else:
logger.debug("SearchResultsProxyModel: Skipping dataChanged emission, model is not active.")
class ThumbnailWorkerSignals(QObject):
"""Classe pour regrouper les signaux du ThumbnailWorker."""
thumbnailReady = pyqtSignal(str, QPixmap, bool) # Chemin du fichier, vignette, succès
finished = pyqtSignal()
errorOccurred = pyqtSignal(str, Exception) # Chemin du fichier, exception
class ThumbnailWorker(QRunnable):
"""Worker pour générer les vignettes de manière asynchrone."""
def __init__(self, file_path, cache):
super().__init__()
self.signals = ThumbnailWorkerSignals() # Instance pour gérer les signaux
self.file_path = file_path
self.cache = cache # Référence au cache partagé
self._cancel_requested = False
@pyqtSlot()
def cancel(self):
"""Annule la génération de la vignette si elle est en cours."""
self._cancel_requested = True
@pyqtSlot() # Supprimer l'argument file_path
def run(self): # Méthode exécutée dans un thread séparé pour générer la vignette
#logger.debug(f"ThumbnailWorker: run() Démarrage pour {self.file_path}")
if self._cancel_requested:
logger.warning(f"ThumbnailWorker: run() cancelled for {self.file_path}")
return
# Vérifier si la vignette est déjà dans le cache
if self.file_path in self.cache:
#logger.debug(f"ThumbnailWorker: Thumbnail found in cache for {self.file_path}")
self.signals.thumbnailReady.emit(self.file_path, self.cache[self.file_path], True)
self.signals.finished.emit()
return
try:
#logger.debug(f"ThumbnailWorker: Generating thumbnail for {self.file_path}")
thumbnail = self.generate_thumbnail(self.file_path)
success = thumbnail is not None and not thumbnail.isNull()
#logger.debug(f"ThumbnailWorker: Thumbnail generated for {self.file_path}, success: {success}")
if success:
self.signals.thumbnailReady.emit(self.file_path, thumbnail, success)
except (FileNotFoundError, ValueError, CancelledError) as e:
logger.error(f"ThumbnailWorker: Error generating thumbnail for {self.file_path}: {e}")
self.signals.errorOccurred.emit(self.file_path, e)
finally:
self.signals.finished.emit()
def generate_thumbnail(self, file_path):
"""Génère une vignette pour le fichier donné."""
#logger.debug(f"ThumbnailWorker: generate_thumbnail called for {file_path}")
try:
if file_path.lower().endswith(('.png', '.jpg', '.jpeg')):
thumbnail = self.generate_image_thumbnail(file_path)
elif file_path.lower().endswith('.dxf'):
thumbnail = self.generate_dxf_thumbnail(file_path)
else:
logger.warning(f"ThumbnailWorker: generate_thumbnail() - Unsupported file format for {file_path}")
raise ValueError(f"Unsupported file format: {file_path}")
return thumbnail
except Exception as e:
logger.error(f"ThumbnailWorker: Error generating thumbnail for {file_path}: {e}", exc_info=True)
raise
def generate_image_thumbnail(self, image_path):
"""Génère une miniature à partir d'un fichier image."""
#logger.debug(f"ThumbnailWorker: generate_image_thumbnail for image: {image_path}")
try:
if self._cancel_requested:
logger.warning(f"ThumbnailWorker: generate_image_thumbnail - Annulé pour {image_path}")
return None # Return None if cancelle
size = QSize(160, 120)
reader = QImageReader(image_path)
reader.setAutoTransform(True)
image = reader.read()
if image.isNull():
logger.error(f"Impossible de lire l'image: {image_path}")
return QPixmap()
thumbnail = QPixmap.fromImage(image).scaled(size, Qt.KeepAspectRatio, Qt.SmoothTransformation)
return thumbnail
except ValueError as ve:
logger.error(f"ValueError creating thumbnail for {image_path}: {ve}", exc_info=True)
return QPixmap()
except Exception as e:
logger.error(f"Error creating thumbnail for {image_path}: {e}", exc_info=True)
return QPixmap() # Return an empty QPixmap on failure
def generate_dxf_thumbnail(self, dxf_file):
"""Génère une miniature à partir d'un fichier DXF."""
#logger.debug(f"ThumbnailWorker: generate_dxf_thumbnail Début de la génération de la miniature DXF pour {self.file_path}")
painter = None # Initialisation
try:
# Vérification d'annulation avant de commencer le traitement
if self._cancel_requested:
logger.warning(f"ThumbnailWorker: generate_dxf_thumbnail - Annulé pour {self.file_path}")
return None
doc = ezdxf.readfile(dxf_file)
msp = doc.modelspace()
# Log the entity types
#entity_types = set(entity.dxftype() for entity in msp)
#logger.info(f"Entity types in DXF file: {entity_types}")
width_thumbnail, height_thumbnail = 120, 160
thumbnail = QPixmap(width_thumbnail, height_thumbnail)
thumbnail.fill(Qt.white)
painter = QPainter(thumbnail)
painter.setRenderHint(QPainter.Antialiasing)
bounding_rect = self.calculate_bounding_rect(msp)
#logger.debug(f"Bounding rect: {bounding_rect}")
scale_factor, offset_x, offset_y = self.calculate_scale_factor_and_offset(bounding_rect, width_thumbnail, height_thumbnail)
#logger.debug(f"Scale factor: {scale_factor}, Offset X: {offset_x}, Offset Y: {offset_y}")
# Vérification d'annulation pendant le traitement
if self._cancel_requested:
logger.warning(f"ThumbnailWorker: generate_dxf_thumbnail - Annulé pendant le traitement de {self.file_path}")
return None
for ent in msp:
if ent.dxftype() == 'LINE':
self.add_line(painter, ent.dxf.start, ent.dxf.end, scale_factor, offset_x, offset_y)
elif ent.dxftype() == 'POINT':
# Handle points if needed
pass
elif ent.dxftype() == 'CIRCLE':
self.add_circle(painter, ent.dxf.center, ent.dxf.radius, scale_factor, offset_x, offset_y)
elif ent.dxftype() == 'ARC':
self.add_arc(painter, ent.dxf.center, ent.dxf.radius, ent.dxf.start_angle, ent.dxf.end_angle, scale_factor, offset_x, offset_y)
elif ent.dxftype() == 'SPLINE':
self.add_spline(painter, ent, scale_factor, offset_x, offset_y)
elif ent.dxftype() == 'ELLIPSE':
self.add_ellipse(painter, ent, scale_factor, offset_x, offset_y)
elif ent.dxftype() == 'TEXT':
self.add_text(painter, ent, scale_factor, offset_x, offset_y)
elif ent.dxftype() == 'LWPOLYLINE':
self.add_wpolyline(painter, ent, scale_factor, offset_x, offset_y)
painter.end()
thumbnail = thumbnail.transformed(QTransform().rotate(-90))
final_thumbnail = thumbnail.scaled(QSize(160, 120), Qt.KeepAspectRatio, Qt.SmoothTransformation)
#logger.debug(f"ThumbnailWorker: Fin de la génération de la miniature DXF pour {self.file_path}")
return final_thumbnail
#return thumbnail
except (ezdxf.DXFError, RecursionError) as e:
# Gérer les erreurs spécifiques au format DXF et les erreurs de récursion
logger.error(f"Erreur DXF ou RecursionError: {e}")
except Exception as e: # Catch-all pour les autres exceptions
logger.error(f"Unexpected error: {type(e).__name__} - {e}", exc_info=True) # Logging avec traceback complet
finally:
if painter and painter.isActive():
painter.end() # Ensure QPainter is ended in case of error
return QPixmap()
def calculate_bounding_rect(self, msp):
def get_arc_extents(center, radius, start_angle, end_angle):
points = []
# Ensure angles are in the range [0, 360)
for angle in [0, 90, 180, 270]:
if start_angle <= angle <= end_angle:
rad_angle = math.radians(angle)
x = center.x + radius * math.cos(rad_angle)
y = center.y + radius * math.sin(rad_angle)
points.append((x, y))
# Add the start and end points of the arc
start_rad = math.radians(start_angle)
end_rad = math.radians(end_angle)
points.append((center.x + radius * math.cos(start_rad), center.y + radius * math.sin(start_rad)))
points.append((center.x + radius * math.cos(end_rad), center.y + radius * math.sin(end_rad)))
min_x = min(point[0] for point in points)
max_x = max(point[0] for point in points)
min_y = min(point[1] for point in points)
max_y = max(point[1] for point in points)
return min_x, max_x, min_y, max_y
min_coord_x, min_coord_y = float('inf'), float('inf')
max_coord_x, max_coord_y = float('-inf'), float('-inf')
for entity in msp:
entity_type = entity.dxftype()
if entity_type == 'LINE':
#logger.debug(f"Processing LINE entity: start={entity.dxf.start}, end={entity.dxf.end}")
min_coord_x = min(min_coord_x, entity.dxf.start.x, entity.dxf.end.x)
max_coord_x = max(max_coord_x, entity.dxf.start.x, entity.dxf.end.x)
min_coord_y = min(min_coord_y, entity.dxf.start.y, entity.dxf.end.y)
max_coord_y = max(max_coord_y, entity.dxf.start.y, entity.dxf.end.y)
elif entity_type == 'CIRCLE':
#logger.debug(f"Processing CIRCLE entity: center={entity.dxf.center}, radius={entity.dxf.radius}")
center = entity.dxf.center
radius = entity.dxf.radius
min_coord_x = min(min_coord_x, center.x - radius, center.x + radius)
max_coord_x = max(max_coord_x, center.x - radius, center.x + radius)
min_coord_y = min(min_coord_y, center.y - radius, center.y + radius)
max_coord_y = max(max_coord_y, center.y - radius, center.y + radius)
elif entity_type == 'ARC':
#logger.debug(f"Processing ARC entity: center={entity.dxf.center}, radius={entity.dxf.radius}, start_angle={entity.dxf.start_angle}, end_angle={entity.dxf.end_angle}")
center = entity.dxf.center
radius = entity.dxf.radius
start_angle = entity.dxf.start_angle
end_angle = entity.dxf.end_angle
min_x, max_x, min_y, max_y = get_arc_extents(center, radius, start_angle, end_angle)
min_coord_x = min(min_coord_x, min_x)
max_coord_x = max(max_coord_x, max_x)
min_coord_y = min(min_coord_y, min_y)
max_coord_y = max(max_coord_y, max_y)
elif entity_type == 'SPLINE':
if hasattr(entity, 'control_points') and entity.control_points:
for point in entity.control_points:
#logger.info(f"Control point: ({point[0]}, {point[1]})")
min_coord_x = min(min_coord_x, point[0])
max_coord_x = max(max_coord_x, point[0])
min_coord_y = min(min_coord_y, point[1])
max_coord_y = max(max_coord_y, point[1])
#else:
#pass
#logger.error(f"Warning: SPLINE entity has no control points. Entity: {entity}")
elif entity_type == 'POINT':
min_coord_x = min(min_coord_x, entity.dxf.location.x)
max_coord_x = max(max_coord_x, entity.dxf.location.x)
min_coord_y = min(min_coord_y, entity.dxf.location.y)
max_coord_y = max(max_coord_y, entity.dxf.location.y)
elif entity_type == 'ELLIPSE':
center = entity.dxf.center
major_axis = entity.dxf.major_axis
ratio = entity.dxf.ratio
# Calcul des points extrêmes de l'ellipse pivotée
points = []
for angle in [0, 90, 180, 270]:
rad_angle = math.radians(angle)
radius = major_axis.magnitude * ratio if abs(math.cos(rad_angle)) > abs(math.sin(rad_angle)) else major_axis.magnitude
x = center.x + radius * math.cos(rad_angle)
y = center.y + radius * math.sin(rad_angle)
points.append((x, y))
min_x = min(point[0] for point in points)
max_x = max(point[0] for point in points)
min_y = min(point[1] for point in points)
max_y = max(point[1] for point in points)
min_coord_x = min(min_coord_x, min_x)
max_coord_x = max(max_coord_x, max_x)
min_coord_y = min(min_coord_y, min_y)
max_coord_y = max(max_coord_y, max_y)
elif entity_type == 'LWPOLYLINE':
#logger.debug(f"Processing LWPOLYLINE entity with {len(entity)} points.")
for point in entity:
#logger.debug(f"Point: {point}")
min_coord_x = min(min_coord_x, point[0])
max_coord_x = max(max_coord_x, point[0])
min_coord_y = min(min_coord_y, point[1])
max_coord_y = max(max_coord_y, point[1])
#logger.debug(f"Updated bounds for LWPOLYLINE: min_x={min_coord_x}, max_x={max_coord_x}, min_y={min_coord_y}, max_y={max_coord_y}")
if min_coord_x == float('inf') or min_coord_y == float('inf'):
logger.debug("Bounding rect has zero width or height. Returning default values.")
return QRectF(0, 0, 0, 0)
#logger.debug(f"Bounding rect: ({min_coord_x}, {min_coord_y}, {max_coord_x}, {max_coord_y})")
return QRectF(min_coord_x, min_coord_y, max_coord_x - min_coord_x, max_coord_y - min_coord_y)
def calculate_scale_factor_and_offset(self, bounding_rect, frame_width, frame_height):
if bounding_rect.width() == 0 or bounding_rect.height() == 0:
logger.warning("Bounding rect has zero width or height. Returning default values to avoid division by zero.")
return 1, 0, 0 # Return default values to avoid division by zero
drawing_width = bounding_rect.height()
drawing_height = bounding_rect.width()
drawing_center_x = (bounding_rect.left() + bounding_rect.right()) / 2
drawing_center_y = (bounding_rect.top() + bounding_rect.bottom()) / 2
scale_factor_width = frame_width / drawing_width
scale_factor_height = frame_height / drawing_height
scale_factor = min(scale_factor_width, scale_factor_height)
offset_x = (frame_width / 2) - (drawing_center_y * scale_factor)
offset_y = (frame_height / 2) - (drawing_center_x * scale_factor)
#logger.debug(f"Scale factor: {scale_factor}, Offset X: {offset_x}, Offset Y: {offset_y}")
return scale_factor, offset_x, offset_y
def add_line(self, painter, start, end, scale_factor, offset_x, offset_y):
#logger.debug(f"Starting to draw line from start={start} to end={end}")
start_point = QPointF((start.y * scale_factor) + offset_x, (start.x * scale_factor) + offset_y)
end_point = QPointF((end.y * scale_factor) + offset_x, (end.x * scale_factor) + offset_y)
#logger.debug(f"Adjusted start_point={start_point}, end_point={end_point}")
painter.drawLine(start_point, end_point)
#logger.debug(f"Finished drawing line: start={start_point}, end={end_point}")
def add_circle(self, painter, center, radius, scale_factor, offset_x, offset_y):
#logger.debug(f"Starting to draw circle with center={center}, radius={radius}")
center_point = QPointF((center.y * scale_factor) + offset_x, (center.x * scale_factor) + offset_y)
scaled_radius = radius * scale_factor
#logger.debug(f"Adjusted center_point={center_point}, scaled_radius={scaled_radius}")
#logger.debug(f"Drawing circle: center={center_point}, scaled_radius={scaled_radius}")
painter.drawEllipse(center_point, scaled_radius, scaled_radius)
#logger.debug(f"Finished drawing circle: center={center_point}, scaled_radius={scaled_radius}")
def add_ellipse(self, painter, ellipse, scale_factor, offset_x, offset_y):
#logger.debug(f"Starting to draw ellipse with center={ellipse.dxf.center}, major_axis={ellipse.dxf.major_axis}, ratio={ellipse.dxf.ratio}")
try:
# Validate inputs
if ellipse.dxf.major_axis.magnitude <= 0:
raise ValueError("Major axis magnitude must be positive")
# The major_axis in DXF represents the vector from the center to the endpoint of the major axis.
# Therefore, the full length of the major axis is twice its magnitude.
major_axis_length = 2*ellipse.dxf.major_axis.magnitude * scale_factor
minor_axis_length = major_axis_length * ellipse.dxf.ratio # Ratio = minor_axis / major_axis
#logger.debug(f"Adjusted major_axis_length={major_axis_length}, minor_axis_length={minor_axis_length}")
# Swap x and y in the center coordinates to match the coordinate system used in other entities
adjusted_center = QPointF(
ellipse.dxf.center.y * scale_factor + offset_x,
ellipse.dxf.center.x * scale_factor + offset_y
)
#logger.debug(f"Adjusted center={adjusted_center}")
# Calculate the rotation angle
major_axis_vector = ellipse.dxf.major_axis
# Swap x and y in the major axis vector components for consistent rotation calculation
scaled_major_axis_x = major_axis_vector.y * scale_factor
scaled_major_axis_y = major_axis_vector.x * scale_factor
rotation_angle = math.degrees(math.atan2(scaled_major_axis_y, scaled_major_axis_x))
#logger.debug(f"Rotation angle={rotation_angle}")
# Save the painter's state, apply transformations, and draw the ellipse
painter.save()
painter.translate(adjusted_center)
painter.rotate(rotation_angle)
rect = QRectF(
-major_axis_length / 2,
-minor_axis_length / 2,
major_axis_length,
minor_axis_length
)
#logger.debug(f"Ellipse bounding rect={rect}")
painter.drawEllipse(rect)
painter.restore()
#logger.debug(f"Drawing ellipse: center={adjusted_center}, major_axis_length={major_axis_length}, minor_axis_length={minor_axis_length}")
except Exception as e:
logger.error(f"Failed to draw ellipse: {e}")
def add_arc(self, painter, center, radius, start_angle, end_angle, scale_factor, offset_x, offset_y):
#logger.debug(f"Original arc parameters: center={center}, radius={radius}, start_angle={start_angle}, end_angle={end_angle}")
start_angle += -90
end_angle += -90
start_angle = start_angle % 360
end_angle = end_angle % 360
start_angle_qt = int(start_angle * 16)
end_angle_qt = int(end_angle * 16)
span_angle_qt = end_angle_qt - start_angle_qt
if span_angle_qt < 0:
span_angle_qt += 360 * 16
center_adjusted = QPointF(center.y * scale_factor + offset_x, center.x * scale_factor + offset_y)
radius_adjusted = radius * scale_factor
#logger.debug(f"Transformed arc parameters: center_adjusted={center_adjusted}, radius_adjusted={radius_adjusted}")
rect = QRectF(center_adjusted.x() - radius_adjusted, center_adjusted.y() - radius_adjusted,
2 * radius_adjusted, 2 * radius_adjusted)
painter.drawArc(rect.toRect(), start_angle_qt, span_angle_qt)
def add_point(self, painter, location, scale_factor, offset_x, offset_y):
point = QPointF(location.y * scale_factor + offset_x, location.x * scale_factor + offset_y)
painter.drawPoint(point)
def add_spline(self, painter, spline, scale_factor, offset_x, offset_y):
#logger.debug("Adding spline with control points: {}".format(spline.control_points))
# Check if the control_points attribute is available and has the expected structure
if hasattr(spline, 'control_points') and spline.control_points:
# Calculate the adjusted control points
spline_control_points = [(point[1] * scale_factor + offset_x, point[0] * scale_factor + offset_y) for point in spline.control_points]
# Reuse the same QPainterPath object
if not hasattr(self, 'spline_painter_path'):
self.spline_painter_path = QPainterPath()
else:
self.spline_painter_path.clear()
# Draw the spline
self.spline_painter_path.moveTo(*spline_control_points[0])
for point in spline_control_points[1:]:
self.spline_painter_path.lineTo(*point)
painter.drawPath(self.spline_painter_path)
else:
logger.error("Warning: SPLINE entity has no control points or the control points have an unexpected structure. Entity: {}".format(spline))
def add_text(self, painter, text_entity, scale_factor, offset_x, offset_y):
text = text_entity.dxf.text
# Adjust text insertion point
insertion_point_adjusted = QPointF(text_entity.dxf.insert.y * scale_factor + offset_x, text_entity.dxf.insert.x * scale_factor + offset_y)
painter.drawText(insertion_point_adjusted, text)
def add_wpolyline(self, painter, lwpolyline, scale_factor, offset_x, offset_y):
"""Dessine une LWPOLYLINE sur le QPainter."""
# Créez un QPainterPath pour dessiner la polyligne
path = QPainterPath()
# Obtenez les points de la LWPOLYLINE
points = lwpolyline.get_points()
# Si la polyligne est fermée, reliez le dernier point au premier
is_closed = lwpolyline.is_closed
if not points:
logger.warning("LWPOLYLINE entity has no points.")
return
# Commencez le chemin au premier point
first_point = points[0]
start_point = QPointF((first_point[1] * scale_factor) + offset_x, (first_point[0] * scale_factor) + offset_y)
path.moveTo(start_point)
# Ajoutez des lignes à chaque point suivant
for point in points[1:]:
next_point = QPointF((point[1] * scale_factor) + offset_x, (point[0] * scale_factor) + offset_y)
path.lineTo(next_point)
# Fermez le chemin si nécessaire
if is_closed:
path.closeSubpath()
# Dessinez le chemin sur le painter
painter.drawPath(path)
class FileViewer(QMainWindow):
folderChanged = pyqtSignal(str)
fileSelected = pyqtSignal(str)
rechercheChanged = pyqtSignal(str)
def __init__(self):
super().__init__()
self.scale_factor = 1 # Initialiser avec une valeur par défaut
self.indexeur = FileIndexer()
self.setWindowTitle("Souf Dxf Explorer")
self.setGeometry(100, 100, 800, 600)
self.model = CustomFileSystemModelForList(self)
self.isSearchThreadRunning = False
self._in_search_mode = False
# Search indicator
self.search_indicator = QLabel("Recherche en cours...", self)
self.statusBar().addWidget(self.search_indicator)
self.search_indicator.hide()
self.current_folder = os.path.normpath(os.path.expanduser("~"))
self.previous_folder = None
#self.current_folders = [QDir.homePath()] # Initialize the current_folders attribute
#self.rootPaths = ["Y:/"]
#logger.debug(f"[Init] Current folders initialized to: {self.current_folders}")
self.setupModelsAndView()
self.setupUIComponents()
self.setupObservers()
self.connect_signals()
self.indexeur.indexUpdated.connect(self.on_indexing_finished)
self.indexeur.indexerError.connect(self.on_indexing_error) # connecter au slot
self.indexeur.indexingProgress.connect(self.update_indexing_progress) # Connecter le signal
# Démarrer l'indexation asynchrone après l'initialisation de l'interface
#QTimer.singleShot(3000, lambda: (self.indexeur.buildIndexAsync(), logger.info(" Fileviewer.__init__(self) (self.indexeur.buildIndexAsync())Lambda function in QTimer.singleShot executed")))
#logger.info("FileViewer / Asynchronous indexing initiated.")
def connect_signals(self):
if not hasattr(self, 'signals_connected'):
# Connexions pour la mise à jour de la vue de la liste de fichiers
self.file_list.model().dataChanged.connect(self.update_file_list_view)
self.model.modelUpdated.connect(self.update_file_list_view)
logger.info("Connection: file_list.clicked -> display_selected_file")
self.file_list.clicked.connect(self.display_selected_file)
logger.info("Connection: file_list.doubleClicked -> on_folder_double_clicked")
self.file_list.doubleClicked.connect(self.on_folder_double_clicked)
# Connexions pour la recherche
self.SearchWorker.searchFinished.connect(self.onSearchFinished)
self.searchThread.started.connect(self.SearchWorker.performSearch)
# Connexions pour les temporisateurs et les champs de texte
self.search_timer.timeout.connect(self.launchSearch)
self.search_box.textChanged.connect(self.launch_delayed_search)
# Configuration du proxy de recherche
self.search_results_proxy_model = SearchResultsProxyModel(self)
self.search_results_proxy_model.setSourceModel(self.model) # Définir le modèle source du proxy
self.signals_connected = True
def setupUIComponents(self):
self.pathLabel = QLabel("Path:")
self.setup_ui()
self.setupMenu()
self.setupToolbar()
def setupModelsAndView(self):
#self.indexeur = FileIndexer()
self.rootPaths = [os.path.normpath("")]
for path in self.rootPaths:
if os.path.exists(path):
print(f"Le chemin {path} est valide.")
else:
#print(f"Le chemin {path} n'est pas valide.")
print(f"Le chemin path est:{path} ")
self.tree_model = CustomFileSystemModelForTree()
self.file_list_model = CustomFileSystemModelForList(self, self.indexeur)
self.file_list = QListView()
self.setup_tree_dock_widget()
self.setup_proxy_tree_model()
# Add SearchResultsProxyModel
self.search_results_proxy_model = SearchResultsProxyModel(self)
self.search_results_proxy_model.setSourceModel(self.file_list_model)
# Connect signal modelReset
self.search_results_proxy_model.modelReset.connect(self.file_list.reset)
# Connect signals layoutChanged and dataChanged
self.search_results_proxy_model.layoutChanged.connect(self.file_list.reset)
self.search_results_proxy_model.dataChanged.connect(self.file_list.reset)
# Configure FileListProxyModel to use SearchResultsProxyModel as source
self.file_list_proxy = FileListProxyModel()
self.setup_file_list_proxy()
self.file_list_proxy.setSourceModel(self.search_results_proxy_model)
# Configure list view to use the proxy file_list_proxy
self.file_list.setModel(self.file_list_proxy)
self.file_list_model.directoryLoaded.connect(self.file_list_model.on_directory_loaded)
def setupObservers(self):
self.SearchWorker = SearchWorker(self.indexeur, self.indexeur.index_dir)
self.searchThread = QThread()
self.SearchWorker.moveToThread(self.searchThread)
self.FileMonitor = FileMonitor(self.indexeur, self.SearchWorker.allowed_extensions)
self.observer = Observer()
self.observer.schedule(self.FileMonitor, path=os.path.normpath(WATCHED_DIR), recursive=True) # Surveiller WATCHED_DIR
self.observer.start()
self.search_timer = QTimer(self)
self.search_timer.setSingleShot(True)
self.update_directory_label(f"Current folder path: {self.tree_model.rootPath()}")
def displayResults(self, success, results):
logger.debug("Beginning of displayResults function.")
# Arrêter le profilage et afficher les résultats
#if hasattr(self, 'profiler'): # Vérifier si le profilage a été activé
#self.profiler.disable()
#s = io.StringIO()
#ps = pstats.Stats(self.profiler, stream=s).sort_stats('cumulative')
#ps.print_stats()
# logger.debug(f"Profiling results:\n{s.getvalue()}")
# Eviter les appels récursifs ou redondants
if hasattr(self, 'display_results_called') and self.display_results_called:
logger.debug("displayResults already called, skipping.")
return
self.display_results_called = True
self.isSearchThreadRunning = False
self.search_indicator.hide()
# Gérer les différents cas de figure
if not success:
logger.debug("Handling search failure.")
self.handleSearchFailure()
self._in_search_mode = False # Assurez de quitter le mode recherche en cas d'échec
elif not results:
logger.debug("Handling no results.")
self.handleNoResults()
self._in_search_mode = False
elif not self.search_box.text().strip():
logger.debug("Search box is empty, resetting list view.")
self.resetListView()
self._in_search_mode = False
else:
# Afficher les résultats de la recherche
self._in_search_mode = True
logger.debug("Updating proxy model with search results.")
self.search_results_proxy_model.setSearchResults(results) # Mettre à jour le proxy
# Si le modèle actuel de la vue n'est pas le modèle de résultats, le définir
if self.file_list.model() != self.search_results_proxy_model:
self.file_list.setModel(self.search_results_proxy_model)
logger.debug("SearchResultsProxyModel set as the current model for file_list.")
# Rafraîchir la vue
#self.file_list.reset()
#logger.debug(f"displayResults: ListView reset. Number of items in view: {self.file_list.model().rowCount()} ")
# Forcer le focus et mettre à jour la vue
#self.file_list.setFocus()
#self.file_list.update()
#self.file_list.repaint()
# (Facultatif) Sélectionner le premier élément si souhaité
if results:
first_result_index = self.search_results_proxy_model.index(0, 0)
if first_result_index.isValid():
self.file_list.setCurrentIndex(first_result_index)
logger.debug("Logging search results.")
self.logSearchResults(results)
self.display_results_called = False
logger.debug("End of displayResults function.")
def launch_delayed_search(self):
logger.debug("Slot activated: launch_delayed_search")
search_criteria = self.search_box.text().strip()
if search_criteria:
logger.info("Search criteria modified. Waiting before launching search...")
if not self.previous_folder:
self.previous_folder = self.current_folder # Sauvegarder le dossier actuel
self.search_timer.stop() # Stop the existing timer before restarting
if self.search_timer.isActive():
self.search_timer.stop()
self.search_timer.start(1000) # Delay of 1000 ms
else:
logger.info("Search bar empty. Displaying current folder.")
self.exit_search_mode()
logger.debug("End of launch_delayed_search")
def launchSearch(self):
logger.debug("Function launchSearch start")
criteria = self.search_box.text().strip()
criteria_type = self.search_criteria_type.currentText().lower().replace(" ", "_")
logger.debug(f"Search criteria: {criteria}, criteria_type: {criteria_type}")
logger.debug(f"Current root path: {self.file_list_model.rootPath()}")
logger.debug(f"WATCHED_DIR: {WATCHED_DIR}")
if not criteria and criteria_type == "filename":
logger.debug("Search criteria is empty, exiting search.")
#QMessageBox.warning(self, "Warning", "Search criteria cannot be empty for filename search.")
self.exit_search_mode()
return
if self.isSearchThreadRunning:
logger.info("Search already running, stopping current search.")
self.SearchWorker.running = False # Stop the thread gracefully
self.searchThread.quit()
self.searchThread.wait()
# Sauvegarder le dossier actuel avant de lancer la recherche
if not self.previous_folder:
self.previous_folder = self.current_folder
logger.debug(f"Saved current folder to previous_folder: {self.previous_folder}")
self.isSearchThreadRunning = True
self.SearchWorker.setCriteria(criteria, criteria_type)
self.SearchWorker.setDateRange(self.start_date.date(), self.end_date.date())
# Assurez-vous que rootPath est bien défini
current_root_path = self.file_list_model.rootPath()
if current_root_path:
self.SearchWorker.setBasePath(current_root_path)
#self.profiler = cProfile.Profile()
#self.profiler.enable()
#self.searchThread.started.connect(self.SearchWorker.performSearch)
self.searchThread.start()
self.search_indicator.show()
logger.debug("Function launchSearch end")
else:
logger.error("Current root path is not set. Cannot start search.")
self.isSearchThreadRunning = False
def onSearchFinished(self, success, results):
logger.debug("Search finished with success: {}".format(success))
self.isSearchThreadRunning = False
if self.searchThread.isRunning():
logger.debug("Stopping search thread.")
self.searchThread.quit()
self.searchThread.wait() # Ensure the thread is fully stopped
logger.debug("Search thread stopped and resources cleaned up.")
self.handle_search_completion(success, results)
def handle_search_completion(self, success, results):
logger.debug("Function handle_search_completion start")
if success:
logger.info("Search completed successfully.")
self.displayResults(success, results)
# Connecter le proxy de recherche à la vue SI en mode recherche
if self._in_search_mode:
self.file_list.setModel(self.search_results_proxy_model)
else:
# Revenir au proxy normal si on n'est plus en recherche
self.file_list.setModel(self.file_list_proxy)
else:
logger.error("Search failed, attempting to restart.")
logger.debug(f"Search results (should be empty on failure): {results}")
# Resetting the thread to allow for new searches
self.searchThread.quit()
self.searchThread.wait() # Wait until the thread cleanly finishes
self.searchThread = QThread() # Recreate the thread for future use
self.SearchWorker = SearchWorker(self.indexeur, self.indexeur.index_dir)
self.SearchWorker.moveToThread(self.searchThread)
self.searchThread.started.connect(self.SearchWorker.performSearch)
self.SearchWorker.searchFinished.connect(self.onSearchFinished)
logger.debug("Function handle_search_completion end")
def handleSearchFailure(self):
logger.debug("Beginning of handleSearchFailure function.")
QMessageBox.warning(self, "Search Failed", "The search failed or was stopped without results.")
self.resetListView()
logger.debug("End of handleSearchFailure function.")
def handleNoResults(self):
logger.debug("Beginning of handleNoResults function.")
QMessageBox.information(self, "No Results", "No results to display.")
self.resetListView()
logger.debug("End of handleNoResults function.")
def logSearchResults(self, results):
logger.debug("Beginning of logSearchResults function.")
for result in results[:min(5, len(results))]:
logger.debug(f"Search result: {result}")
logger.debug("End of logSearchResults function.")
def exit_search_mode(self):
logger.debug("Function exit_search_mode start")
self._in_search_mode = False
self.file_list_proxy.invalidateFilter()
if self.previous_folder:
logger.debug(f"[exit_search_mode] Previous folder: {self.previous_folder}")
self.current_folder = self.previous_folder # Restaurer le dossier précédent
self.previous_folder = None # Réinitialiser previous_folder
self.file_list_model.setRootPath(self.current_folder)
root_index = self.file_list_model.indexFromPath(self.current_folder)
if root_index.isValid():
if hasattr(self, 'file_list_proxy'):
self.file_list_proxy.setSourceModel(self.file_list_model)
self.file_list.setModel(self.file_list_proxy)
proxy_root_index = self.file_list_proxy.mapFromSource(root_index)
if proxy_root_index.isValid():
self.file_list.setRootIndex(proxy_root_index)
else:
logger.error(f"Invalid proxy root index: {proxy_root_index}")
else:
self.file_list.setModel(self.file_list_model)
self.file_list.setRootIndex(root_index)
else:
logger.error(f"Invalid root index: {root_index}")
self.update_directory_label(f"Current folder: {self.current_folder}")
logger.info(f"[exit_search_mode] Current folder updated to: {self.current_folder}")
else:
logger.warning("[exit_search_mode] No previous folder stored. Using default path.")
default_folder = QDir.homePath()
self.file_list_model.setRootPath(default_folder)
root_index = self.file_list_model.index(default_folder)
if hasattr(self, 'file_list_proxy'):
self.file_list_proxy.setSourceModel(self.file_list_model)
proxy_root_index = self.file_list_proxy.mapFromSource(root_index)
self.file_list.setRootIndex(proxy_root_index)
else:
self.file_list.setRootIndex(root_index)
self.update_directory_label(f"Current folder: {default_folder}")
logger.info(f"[Update] Current folder updated to: {default_folder}")
logger.debug("Function exit_search_mode end")
def resetListView(self):
"""Réinitialise la vue de liste pour afficher le contenu du répertoire courant."""
try:
self._in_search_mode = False # Sortir du mode recherche
self.file_list.setModel(self.file_list_proxy) # Rétablir le proxy principal
# Vérifier si current_folder est valide avant de continuer
if self.current_folder and os.path.exists(self.current_folder):
logger.debug(f"[resetListView] Current folder: {self.current_folder}")
self.updateFileListModel(self.current_folder)
self.update_file_list_view()
self.checkDisplayedFiles(self.file_list_model)
self.updateDirectoryLabel()
else:
self.handleNoCurrentFolder()
self.file_list.reset() # Rafraîchir la vue après les modifications
except Exception as e:
self.handleResetListViewError(e)
def updateFileListModel(self, folder_path):
"""Met à jour le chemin racine du modèle de liste de fichiers.
Args:
folder_path (str): Le nouveau chemin racine pour le modèle.
"""
logger.debug("Fonction updateFileListModel: Début")
if not isinstance(folder_path, str) or not folder_path:
logger.error("Le chemin du dossier doit être une chaîne de caractères valide.")
QMessageBox.warning(self, "Erreur", "Le chemin du dossier doit être une chaîne de caractères valide.")
return
if not os.path.isdir(folder_path):
logger.error(f"Le chemin fourni n'est pas valide ou n'est pas un dossier: {folder_path}")
QMessageBox.warning(self, "Erreur", "Le chemin fourni n'est pas un dossier valide.")
return
self.file_list_model.setRootPath(folder_path)
logger.info(f"Chemin racine du modèle de liste de fichiers mis à jour: {folder_path}")
logger.debug("Fonction updateFileListModel: Fin")
def update_file_list_view(self, folder_path=None, index=None):
logger.debug("Fonction update_file_list_view: Début")
try:
root_index = None
if folder_path is not None:
if isinstance(folder_path, str):
logger.debug(f"Chemin du dossier fourni: {folder_path}")
self.updateFileListModel(folder_path)
root_index = self.file_list_model.indexFromPath(folder_path)
if not root_index.isValid():
logger.error("Index racine obtenu à partir du chemin est invalide.")
return
else:
logger.error("Le chemin du dossier doit être une chaîne de caractères valide.")
return
elif index is not None:
logger.debug(f"Index fourni: {index}")
if not index.isValid():
logger.error("Index non valide reçu pour la mise à jour.")
return
root_index = index if not isinstance(index.model(), QSortFilterProxyModel) else index.model().mapToSource(index)
if not root_index.isValid():
logger.error("Index racine obtenu à partir de l'index fourni est invalide.")
return
else:
if self.current_folder:
folder_path = self.current_folder
logger.debug(f"Utilisation du dossier actuel: {folder_path}")
self.updateFileListModel(folder_path)
root_index = self.file_list_model.indexFromPath(folder_path)
if not root_index.isValid():
logger.error("Index racine obtenu à partir du dossier actuel est invalide.")
return
else:
logger.error("Aucun dossier courant défini.")
return
if hasattr(self, 'file_list_proxy'):
self.file_list_proxy.setSourceModel(self.file_list_model)
proxy_root_index = self.file_list_proxy.mapFromSource(root_index)
if proxy_root_index.isValid():
self.file_list.setRootIndex(proxy_root_index)
logger.info(f"Vue de la liste mise à jour avec proxy. Index source: {root_index}, Index proxy: {proxy_root_index}")
else:
logger.error("Erreur lors de la mise à jour de la vue avec proxy: L'index proxy n'est pas valide.")
else:
if root_index.isValid():
self.file_list.setRootIndex(root_index)
logger.info(f"Vue de la liste mise à jour sans proxy. Index source: {root_index}")
else:
logger.error("Erreur lors de la mise à jour de la vue sans proxy: L'index racine n'est pas valide.")
except Exception as e:
logger.error(f"Erreur dans update_file_list_view: {e}")
QMessageBox.warning(self, "Erreur", f"Une erreur s'est produite: {e}")
logger.debug("Fonction update_file_list_view: Fin")
def showModelInfo(self):
logger.debug("Beginning of showModelInfo function.")
try:
self.logModelContents(self.file_list_model)
except Exception as e:
logger.error(f"An error occurred while displaying model information: {str(e)}")
QMessageBox.critical(self, "Error", f"An error occurred: {str(e)}")
logger.debug("End of showModelInfo function.")
def logModelContents(self, model):
logger.debug("Beginning of logModelContents function.")
root_path = model.rootPath()
root_index = model.indexFromPath(root_path)
for row in range(min(5, model.rowCount(root_index))):
child_index = model.index(row, 0, root_index)
if child_index.isValid():
file_path = model.filePath(child_index)
logger.debug(f"Element at index {row}: {file_path}")
else:
logger.warning(f"Invalid index at row {row}")
logger.debug("End of logModelContents function.")
def handleNoCurrentFolder(self):
logger.info("No current folder set, not updating the file list.")
self.current_folder = os.path.expanduser("~") # Chemin par défaut vers le dossier utilisateur
self.updateFileListModel(self.current_folder)
self.updateDirectoryLabel()
self.update_file_list_view()
# You can add additional logic here, such as displaying a message to the user
def setCurrentFolderPath(self):
default_folder = os.path.expanduser("~") # Ou un autre chemin par défaut
self.current_folder = default_folder
logger.debug(f"Current folder path reset to default: {self.current_folder}")
def updateDirectoryLabel(self):
self.update_directory_label(f"Current folder: {self.current_folder}")
logger.debug("Directory label updated.")
def handleResetListViewError(self, error):
logger.error(f"Error resetting list view: {error}")
QMessageBox.warning(self, "Error", "Problem resetting list view.")
logger.warning("Reset list view error handled.")
def checkDisplayedFiles(self, model):
actual_files = [model.filePath(model.index(row, 0)) for row in range(model.rowCount())]
logger.info("Files currently displayed in the view: " + str(actual_files))
@pyqtSlot(int, int)
def update_indexing_progress(self, processed_files, total_files):
# Mettre à jour la barre de progression ou afficher un message de statut
if total_files > 0:
progress_percent = int(processed_files / total_files * 100)
QMetaObject.invokeMethod(self, "update_status_bar", Qt.QueuedConnection,
Q_ARG(str, f"Indexing in progress... {progress_percent}%"))
else:
progress_percent = 0
@pyqtSlot(str)
def update_status_bar(self, message):
self.statusBar().showMessage(message)
@pyqtSlot()
def on_indexing_finished(self):
"""Slot appelé lorsque l'indexation est terminée."""
logger.info("FileViewer / Indexing finished successfully.")
# Ajoutez ici le code que vous souhaitez exécuter à la fin de l'indexation
#self.update_file_list_view(WATCHED_DIR) # Mettre à jour la vue de liste après l'indexation
# ... (autres actions)
# Activez la recherche lorsque l'indexation est terminée
self.search_button.setEnabled(True)
@pyqtSlot(str)
def on_indexing_error(self, error_message):
"""Slot appelé en cas d'erreur lors de l'indexation."""
logger.error(f"FileViewer / Indexing error: {error_message}")
# Gérez l'erreur ici (par exemple, affichez un message à l'utilisateur)
self.is_indexing = False # Reset the flag in case of an error
QMessageBox.critical(self, "Erreur d'indexation", f"Une erreur est survenue lors de l'indexation : {error_message}")
def get_proxy_model_for_source_model(self, source_model):
logger.debug("Function get_proxy_model_for_source_model start")
if source_model == self.file_list_model:
logger.info("Found matching proxy model for file list model.")
return self.file_list_proxy
elif source_model == self.tree_model:
logger.info("Found matching proxy model for tree model.")
return self.tree_proxy
logger.warning("No matching proxy model found for given source model.")
logger.debug("Function get_proxy_model_for_source_model end")
return None
def get_view_for_model(self, model):
logger.debug("Function get_view_for_model start")
if model == self.file_list_model:
logger.info("Found matching view for file list model.")
return self.file_list
elif model == self.tree_model:
logger.info("Found matching view for tree model.")
return self.tree_view
logger.debug("Function get_view_for_model end")
return None
def on_root_path_changed(self, new_path):
#self.current_folder = new_path
logger.debug("Function on_root_path_changed start with new path: {}".format(new_path))
self.file_list_proxy.setSourceModel(self.file_list_model) # Re-attach source model if needed
self.update_file_list_view()
logger.debug("Function on_root_path_changed end")
def update_directory_label(self, text=None):
logger.info("Function update_directory_label start")
if text is None:
text = f"Current folder: {self.current_folder}"
logger.debug(f"Attempting to update directory label with text: {text}")
if hasattr(self, 'directory_label'):
self.directory_label.setText(text)
logger.info(f"Updated directory label: {text}")
else:
logger.error("Error: Directory label is not initialized.")
# Optionally initialize the label here if needed
self.directory_label = QLabel(text)
logger.info(f"Initialized and updated directory label: {text}")
logger.info("Function update_directory_label end")
def onRecursiveSearchToggled(self, checked):
logger.debug("Function onRecursiveSearchToggled start")
self.file_list_proxy.setRecursiveSearchEnabled(checked)
logger.info(f"Recursive search {'enabled' if checked else 'disabled'}.")
logger.debug("Function onRecursiveSearchToggled end")
def setup_proxy_tree_model(self):
logger.debug("Starting setup of the proxy model for the tree/Function setup_proxy_tree_model")
if self.tree_model is None:
logger.error("Error: The source model (tree_model) is not initialized.")
return
if hasattr(self, 'tree_proxy_initialized') and self.tree_proxy_initialized:
logger.debug("Proxy model for the tree is already set up.")
return
self.tree_proxy = TreeProxyModel()
self.tree_proxy.setSourceModel(self.tree_model)
logger.info("Source model assigned to the tree proxy.")
tree_filters = ['*'] # Specific filters for the tree
self.tree_proxy.setFilterWildcard(' '.join(tree_filters))
self.tree_proxy.invalidateFilter()
logger.info("Filter of the tree proxy model invalidated to reflect the changes.")
self.tree_view.setModel(self.tree_proxy)
self.tree_proxy_initialized = True
logger.debug("End of the setup of the proxy model for the tree/Function setup_proxy_tree_model")
def setup_file_list_proxy(self):
logger.debug("Starting setup of the proxy model for the file list./Function setup_file_list_proxy")
if self.file_list_model is None:
logger.error("Error: The source model (file_list_model) is not initialized.")
return
try:
self.file_list_proxy = FileListProxyModel()
self.file_list_proxy.setSourceModel(self.file_list_model)
logger.info("Source model assigned to the file list proxy.")
# Configuring filters
file_list_filters = ['*.dxf', '*.jpg', '*.jpeg', '*.png', '*.bmp'] # Filters for the file list view
self.file_list_proxy.setFilterWildcard(' '.join(file_list_filters))
logger.info(f"Filters applied for the file list proxy model: {' '.join(file_list_filters)}")
# Applying filters and updating the view
self.file_list_proxy.invalidateFilter()
logger.info("Filter of the file list proxy model invalidated to reflect the changes.")
self.file_list.setModel(self.file_list_proxy)
except Exception as e:
logger.error(f"Error during the configuration of the proxy model: {str(e)}")
logger.debug("End of the setup of the proxy model for the file list./Function setup_file_list_proxy")
def setup_ui(self):
logger.debug("Starting the setup_ui function in FileViewer")
self.setupTreeModelAndView()
self.setupListModelAndView()
self.central_layout = QVBoxLayout()
self.main_splitter = QSplitter(Qt.Horizontal)
self.setCentralWidget(self.main_splitter)
self.tree_dock_widget.setFeatures(QDockWidget.NoDockWidgetFeatures)
self.setup_central_frame()
self.setup_preview_dock_widget()
self.main_splitter.setSizes([445])
logger.info("Connecting signals and slots:")
self.central_layout.addWidget(self.file_list)
self.directory_label = QLabel("Current folder path:")
logger.debug("[Init] Directory label initialized")
self.central_layout.addWidget(self.directory_label)
logger.debug("[Layout] Directory label added to layout")
self.file_list.setContextMenuPolicy(Qt.CustomContextMenu)
self.file_list.customContextMenuRequested.connect(self.open_context_menu)
logger.debug("End of the setup_ui function in FileViewer")
def handle_root_path_changed(self, new_root_path):
logger.debug("Starting the handle_root_path_changed function")
logger.info(f"New root path: {new_root_path}")
current_root_path = self.file_list_model.rootPath()
if new_root_path != current_root_path:
self.current_folder = new_root_path # Mettre à jour current_folder
self.update_file_list_view(new_root_path)
logger.debug("End of the handle_root_path_changed function")
def setupTreeModelAndView(self):
logger.debug("Starting the setupTreeModelAndView function")
self.tree_model = CustomFileSystemModelForTree()
self.tree_model.setRootPath("")
self.tree_proxy = TreeProxyModel()
self.tree_proxy.setSourceModel(self.tree_model)
self.tree_view.setModel(self.tree_proxy)
logger.debug("End of the setupTreeModelAndView function")
def setupListModelAndView(self):
logger.debug("Starting the setupListModelAndView function")
self.file_list_model.setRootPath(self.current_folder)
logger.info(f"Initial root path for the file list model: (empty)")
# Configuring the proxy model
self.file_list_proxy = FileListProxyModel()
self.file_list_proxy.setSourceModel(self.file_list_model)
logger.info("Proxy model configured with the file list model.")
self.file_list_proxy.setFilterKeyColumn(0)
self.file_list_proxy.setFilterWildcard("*.dxf *.jpg *.jpeg *.png *.bmp")
# Associating the proxy model with the view
self.file_list.setModel(self.file_list_proxy)
logger.info("File list view configured with the proxy model.")
self.file_list.setRootIndex(QModelIndex())
logger.info("File list view configured to display empty content.")
#self.showModelInfo()
logger.info("Triggered showModelInfo:")
logger.debug("End of the setupListModelAndView function")
def setup_tree_dock_widget(self):
logger.debug("Start of setup_tree_dock_widget function")
self.tree_dock_widget = QDockWidget("Tree")
self.tree_view = QTreeView(self.tree_dock_widget)
self.tree_dock_widget.setWidget(self.tree_view)
self.tree_dock_widget.setMinimumWidth(455)
self.tree_view.clicked.connect(self.on_tree_clicked)
logger.info("Connection: tree_view.clicked -> on_tree_clicked")
#self.tree_model = CustomFileSystemModelForTree()
self.tree_model.setRootPath("")
root_path = self.tree_model.rootPath()
logger.info(f"Tree root path: {root_path}")
self.setup_proxy_tree_model()
header = self.tree_view.header()
header.setSectionHidden(1, True)
header.setSectionHidden(2, True)
header.setSectionHidden(3, True)
self.tree_view.setHeaderHidden(False) # Hide header if desired
self.addDockWidget(Qt.LeftDockWidgetArea, self.tree_dock_widget)
logger.debug("End of setup_tree_dock_widget function")
def setup_preview_dock_widget(self):
logger.debug("Start of setup_preview_dock_widget function")
self.preview_dock_widget = QDockWidget("Preview")
self.preview_dock_widget.setFixedHeight(455)
self.preview_dock_widget.setMinimumWidth(455)
self.preview_scene = QGraphicsScene()
self.preview_view = CustomGraphicsView(self.preview_scene)
self.preview_dock_widget.setWidget(self.preview_view)
self.addDockWidget(Qt.LeftDockWidgetArea, self.preview_dock_widget)
# Barre d'outils pour le zoom et les modes de mesure
self.toolbar = QToolBar("Zoom and Measure Tools", self)
self.addToolBar(Qt.TopToolBarArea, self.toolbar)
zoom_in_action = QAction("Zoom In", self)
zoom_in_action.triggered.connect(self.preview_view.zoom_in)
self.toolbar.addAction(zoom_in_action)
zoom_out_action = QAction("Zoom Out", self)
zoom_out_action.triggered.connect(self.preview_view.zoom_out)
self.toolbar.addAction(zoom_out_action)
# Séparateur visuel
self.toolbar.addSeparator()
# Actions pour les modes de mesure
action_measure_distance = QAction(QIcon("iconDistance.png"), "Distance Mode", self)
action_measure_distance.triggered.connect(lambda: self.preview_view.set_measure_mode(MeasureMode.DISTANCE))
self.toolbar.addAction(action_measure_distance)
action_measure_angle = QAction(QIcon("iconAngle.png"), "Angle Mode", self)
action_measure_angle.triggered.connect(lambda: self.preview_view.set_measure_mode(MeasureMode.ANGLE))
self.toolbar.addAction(action_measure_angle)
# Configuration de la vue et de la scène
self.canvas = CustomGraphicsView()
self.preview_view.setScene(self.preview_scene)
self.preview_view.fitInView(self.preview_scene.sceneRect(), Qt.KeepAspectRatio)
self.scene = QGraphicsScene()
self.canvas.setScene(self.scene)
logger.debug("End of setup_preview_dock_widget function")
def setup_central_frame(self):
logger.debug("Start of setup_central_frame function")
self.central_frame = QWidget()
self.central_layout = QVBoxLayout(self.central_frame)
self.label_files = QLabel("File List:")
self.central_layout.addWidget(self.label_files)
self.search_box = QLineEdit()
self.search_box.setPlaceholderText("Search files...")
self.central_layout.addWidget(self.search_box)
self.search_criteria_type = QComboBox(self)
self.search_criteria_type.addItems(["Filename", "Creation Date", "Modification Date"])
self.central_layout.addWidget(self.search_criteria_type)
self.start_date = QDateEdit(self)
self.start_date.setCalendarPopup(True)
self.start_date.setDate(QDate.currentDate())
self.central_layout.addWidget(self.start_date)
self.end_date = QDateEdit(self)
self.end_date.setCalendarPopup(True)
self.end_date.setDate(QDate.currentDate())
self.central_layout.addWidget(self.end_date)
self.search_button = QPushButton("Search", self)
self.central_layout.addWidget(self.search_button)
self.file_list = QListView()
self.file_list.setIconSize(QSize(160, 120))
logger.info(f"New icon size for QListView: {self.file_list.iconSize()}")
self.file_list.setGridSize(QSize(175, 135))
self.file_list.setModel(self.file_list_proxy ) # Use the file list proxy model here
self.file_list.setViewMode(QListView.IconMode)
#self.file_list_model.setRootPath(QDir.rootPath())
root_index_model_files = self.file_list_proxy.sourceModel().index(0, 0)
proxy_root_index = self.file_list_proxy.mapFromSource(root_index_model_files)
self.file_list.setRootIndex(proxy_root_index)
self.central_layout.addWidget(self.file_list)
self.main_splitter.addWidget(self.central_frame)
logger.debug("End of setup_central_frame function")
def on_signal_emitted(self, index):
logger.debug(f"Start on_signal_emitted (self.update_file_list(index))")
logger.debug(f"Signal emitted. Index: {index}")
if index.isValid():
folder_path = self.file_list_model.filePath(index)
self.current_folder = folder_path
self.update_file_list_view(folder_path)
else:
logger.error("Invalid index received in on_signal_emitted")
logger.debug(f"End on_signal_emitted")
def on_folder_double_clicked(self, index):
logger.debug("Start on_folder_double_clicked")
if not index.isValid():
logger.error("The index is not valid.")
QMessageBox.warning(self, "Error", "Invalid index selected.")
return
# Détermination du modèle de base et de l'indice source
if isinstance(index.model(), QSortFilterProxyModel):
source_index = index.model().mapToSource(index)
if index.model() == self.file_list_proxy:
base_model = self.file_list_model
logger.info("Index from file list proxy")
elif index.model() == self.tree_proxy:
base_model = self.tree_model
logger.info("Index from tree proxy")
else:
QMessageBox.warning(self, "Error", "Unknown model type")
return
else:
source_index = index
base_model = index.model()
folder_path = base_model.filePath(source_index)
if os.path.isdir(folder_path) and os.access(folder_path, os.R_OK):
logger.info(f"Selected folder path: {folder_path}")
self.current_folder = folder_path
logger.debug(f"[on_folder_double_clicked] Current folder updated to: {self.current_folder}")
self.file_list_model.setRootPath(folder_path)
logger.info(f"Root path of the file list model updated: {folder_path}")
# Mise à jour de l'indice root de la vue
updated_root_index = self.file_list_model.indexFromPath(folder_path)
if hasattr(self, 'file_list_proxy'):
proxy_root_index = self.file_list_proxy.mapFromSource(updated_root_index)
if proxy_root_index.isValid():
self.file_list.setRootIndex(proxy_root_index)
else:
logger.error("Invalid proxy root index after mapping.")
QMessageBox.warning(None, "Mapping Error", "Invalid proxy root index after mapping.")
else:
self.file_list.setRootIndex(updated_root_index)
logger.info("List view updated to display the content of the selected folder.")
self.update_directory_label(f"Current folder path: {self.current_folder}")
else:
msg = "The selected path is not a directory." if not os.path.isdir(folder_path) else "The directory is not accessible."
logger.error(f"{msg}: {folder_path}")
QMessageBox.warning(self, "Error", msg)
logger.debug("End on_folder_double_clicked")
def on_tree_clicked(self, index):
logger.debug("Start of on_tree_clicked")
if self._in_search_mode:
logger.info("Exiting search mode after tree click.")
self._in_search_mode = False
self.file_list.setModel(self.file_list_proxy)
self.search_box.clear()
logger.debug(f"Start processing index: {index}")
if not index.isValid():
logger.error("The provided index is invalid.")
return
# Map the proxy index to the source index
source_index = self.tree_proxy.mapToSource(index) if isinstance(index.model(), QSortFilterProxyModel) else index
if not source_index.isValid():
logger.error("Index mapping failed. Attempting reset.")
return
# Get the directory path
try:
directory_path = self.tree_model.filePath(source_index)
self.current_folder = directory_path
except IndexError:
logger.error(f"Error getting directory path from index: {source_index}")
return
assert os.path.isdir(directory_path), f"Invalid directory path: {directory_path}"
logger.debug(f"Directory path: {directory_path}")
#Only update if the directory is different
if directory_path == self.file_list_model.rootPath():
logger.debug("Selected directory is already the active directory. Skipping update.")
return
# Update the file list model
logger.info(f"Updating file list model for directory: {directory_path}")
self.file_list_model.setRootPath(directory_path)
# Directly update the list view after setting the root path
list_root_index = self.file_list_model.indexFromPath(directory_path)
self.update_view_index(self.file_list_model, self.file_list, list_root_index)
self.update_directory_label(f"Current folder path: {self.current_folder}")
logger.info(f"[on_tree_clicked] Current folder path updated to: {directory_path}")
logger.debug("End of on_tree_clicked")
def update_view_index(self, model, view, root_index):
"""Updates the view's index based on the model and root index."""
if hasattr(self, 'file_list_proxy') and model is self.file_list_model:
index_proxy = self.file_list_proxy.mapFromSource(root_index)
if index_proxy.isValid():
view.setRootIndex(index_proxy)
else:
logger.error("Root proxy index is not valid after mapping.")
else:
view.setRootIndex(root_index)
def load_folder(self):
"""Loads a folder selected by the user."""
logger.info("Start of load_folder function")
folder_path = QFileDialog.getExistingDirectory(self, "Choose a folder")
if folder_path:
try:
# Check if the path is accessible (read and write)
if not os.access(folder_path, os.R_OK | os.W_OK):
raise OSError("The selected folder is not accessible for read or write.")
logger.info(f"Selected folder path: {folder_path}")
self.current_folder = folder_path
self.update_directory_label(f"Current folder path: {self.current_folder}")
logger.debug(f"[load_folder] Current folder updated to: {self.current_folder}")
# Update path in the list model
self.file_list_model.setRootPath(self.current_folder)
# Update list view index
list_root_index = self.file_list_model.indexFromPath(self.current_folder)
self.update_view_index(self.file_list_model, self.file_list, list_root_index)
self.folderChanged.emit(folder_path)
except OSError as e:
QMessageBox.warning(self, "Error", str(e))
logger.error(f"Error accessing folder: {e}")
else:
logger.warning("No folder selected.")
logger.info("End of load_folder function")
def navigate_up(self):
logger.info("Start navigate_up")
logger.debug("Attempting to navigate up one level in the file tree.")
current_index = self.file_list.rootIndex()
index_parent = self.file_list_proxy.mapToSource(current_index).parent() # Use the correct proxy model here
if index_parent.isValid():
self.file_list.setRootIndex(self.file_list_proxy.mapFromSource(index_parent))
folder_path = self.file_list_model.filePath(index_parent)
self.current_folder = folder_path
self.update_directory_label(f"Current folder path: {self.current_folder}")
logger.info(f"Current directory updated to: {self.current_folder}")
else:
logger.warning("Unable to navigate up, no valid parent found.")
logger.info("End navigate_up")
def on_tree_selection_changed(self, selected, deselected):
logger.debug("Function on_tree_selection_changed start")
index = self.tree_view.currentIndex()
logger.debug(f"Current index in the tree: {index}")
self.navigate_folder(index)
logger.debug("Function on_tree_selection_changed end")
def on_list_selection_changed(self, selected, deselected):
logger.debug("Function on_list_selection_changed start")
index_source = self.file_list.currentIndex()
logger.debug(f"Current source index in the list: {index_source}")
self.navigate_folder(index_source)
logger.debug("Function on_list_selection_changed end")
def navigate_folder(self, index):
logger.debug("Function navigate_folder start")
if not index.isValid():
logger.error("Error: Invalid index provided for navigation.")
return
folder_path = self.file_list_model.filePath(index)
if not os.path.exists(folder_path):
logger.error(f"Error: Folder path does not exist - {folder_path}")
return
self.current_folder = folder_path # Mise à jour du dossier courant
self.file_list_model.setRootPath(self.current_folder)
logger.info(f"Updated root path of the file list model to: {self.current_folder}")
self.update_file_list_view(folder_path)
self.update_directory_label(f"Current folder: {self.current_folder}")
logger.debug("Function navigate_folder end")
def change_list_view(self):
logger.info("Start change_list_view")
logger.info("Switching view to List View")
self.file_list.setViewMode(QListView.ListMode)
logger.debug("File view switched to list.")
logger.info("End change_list_view")
def change_thumbnail_view(self):
logger.info("Start change_thumbnail_view")
logger.info("Switching view to Thumbnail View")
self.file_list.setViewMode(QListView.IconMode)
logger.debug("File view switched to thumbnails.")
logger.info("End change_thumbnail_view")
def setupMenu(self):
logger.info("Start setupMenu")
logger.debug("Setting up the menu bar")
menuBar = self.menuBar()
fileMenu = menuBar.addMenu("&File")
openAction = QAction("&Open", self)
openAction.setShortcut("Ctrl+O")
openAction.triggered.connect(self.load_folder) # Assuming `load_folder` is your method to open a folder
exitAction = QAction("&Exit", self)
exitAction.setShortcut("Ctrl+Q")
exitAction.triggered.connect(self.close)
fileMenu.addAction(openAction)
fileMenu.addSeparator()
fileMenu.addAction(exitAction)
viewMenu = menuBar.addMenu("&View")
listViewAction = QAction("&List View", self)
listViewAction.triggered.connect(self.change_list_view)
viewMenu.addAction(listViewAction)
thumbnailViewAction = QAction("&Thumbnail View", self)
thumbnailViewAction.triggered.connect(self.change_thumbnail_view)
viewMenu.addAction(thumbnailViewAction)
logger.debug("Menu bar configured with File and View menus.")
logger.info("End setupMenu")
def setupToolbar(self):
logger.info("Start setupToolbar")
logger.debug("Setting up the toolbar")
toolbar = self.addToolBar("Main Toolbar")
returnAction = QAction(QIcon("iconReturn.png"), "Return", self)
toolbar.setIconSize(QSize(24, 24))
returnAction.triggered.connect(self.navigate_up)
toolbar.addAction(returnAction)
logger.debug("Toolbar configured")
logger.info("End setupToolbar")
def closeEvent(self, event):
logger.info("Function closeEvent start")
logger.info("Closing application, stopping threads and observer.")
try:
if hasattr(self, 'observer') and self.observer:
self.observer.stop()
self.observer.join()
if hasattr(self, 'searchThread') and self.searchThread.isRunning():
self.SearchWorker.running = False # Assurez-vous que SearchWorker peut être arrêté proprement
#self.SearchWorker.stop()
self.searchThread.quit()
self.searchThread.wait() # Wait up to 0 seconds for the thread to cleanly finish
except Exception as e:
logger.error(f"Error during closing: {e}")
finally:
super().closeEvent(event)
logger.info("Function closeEvent end")
def display_file(self, file):
logger.debug(f"Start of display_file function with file: {file}")
try:
if os.path.isdir(file):
self.display_folder(file)
elif file.lower().endswith(('.png', '.jpg', '.jpeg')):
self.display_image(file)
elif file.lower().endswith('.dxf'):
self.display_dxf(file)
else:
self.display_default_message()
except Exception as e:
logger.error(f"Error displaying file {file}: {e}")
self.display_default_message()
logger.info(f"End of display_file function")
def display_folder(self, folder):
logger.info(f" Function display_folder / Displaying folder: {folder}")
self.current_folder = folder # Mise à jour du dossier courant
self.update_directory_label(f"Current folder: {self.current_folder}")
# Ajoutez le code nécessaire pour afficher le contenu du dossier si nécessaire
def display_selected_file(self, index):
"""Affiche le fichier sélectionné dans le dock de prévisualisation."""
logger.info("Start of display_selected_file")
logger.debug(f"Function display_selected_file with index: {index}")
try:
# Déterminer le modèle à utiliser en fonction du mode de recherche
if self._in_search_mode:
model = self.search_results_proxy_model
else:
model = self.file_list_proxy
if not model:
logger.error("Model is not initialized")
return
# Convertir l'index du proxy en index du modèle source
source_index = model.mapToSource(index)
if not source_index.isValid():
logger.error("Invalid index for file selection")
return
file_path = self.file_list_model.filePath(source_index)
if not file_path:
logger.error("File path could not be retrieved")
return
logger.debug(f"Ready to open: {file_path}")
logger.info(f"Displaying file for index {index}: {file_path}")
logger.debug(f"FileViewer.display_selected_file() - Appel de display_file pour {file_path}")
self.display_file(file_path)
self.fileSelected.emit(file_path)
except Exception as e:
logger.error(f"Error in display_selected_file: {e}")
logger.info("End of display_selected_file")
def display_image(self, image_path):
logger.info(f"Start display_image")
pixmap = QPixmap(image_path)
self.preview_scene.clear()
item = QGraphicsPixmapItem(pixmap)
item.setTransform(QTransform())
self.preview_scene.addItem(item)
rectF = QRectF(pixmap.rect())
self.preview_scene.setSceneRect(rectF)
self.preview_view.setTransform(QTransform())
self.preview_view.fitInView(rectF, Qt.KeepAspectRatio)
self.preview_view.viewport().update()
#self.preview_view.update()
logger.info(f"Image displayed in preview: {image_path}")
logger.info(f"End display_image")
def display_dxf(self, file):
logger.debug("Starting function display_dxf")
try:
self.preview_scene.clear()
doc = ezdxf.readfile(file)
msp = doc.modelspace()
window_size = self.preview_view.size()
scale_factor = self.calculate_scale_factor(msp, window_size)
self.scale_factor = scale_factor # Stocker le facteur d'échelle
entities_to_add = [] # Use a list to batch add if possible
for ent in msp:
if ent.dxftype() == 'LINE':
entities_to_add.append(self.add_line(ent, scale_factor))
elif ent.dxftype() == 'CIRCLE':
entities_to_add.append(self.add_circle(ent, scale_factor))
elif ent.dxftype() == 'SPLINE':
entities_to_add.append(self.add_spline(ent, scale_factor))
elif ent.dxftype() == 'POINT':
entities_to_add.append(self.add_point(ent, scale_factor))
elif ent.dxftype() == 'ARC':
entities_to_add.append(self.add_arc(ent, scale_factor))
elif ent.dxftype() == 'ELLIPSE':
entities_to_add.append(self.add_ellipse(ent, scale_factor))
elif ent.dxftype() == 'TEXT':
entities_to_add.append(self.add_text(ent, scale_factor))
#elif ent.dxftype() == 'MTEXT':
#entities_to_add.append(self.add_mtext(ent, scale_factor))
elif ent.dxftype() == '3DFACE':
entities_to_add.append(self.add_3dface(ent, scale_factor))
elif ent.dxftype() == 'SOLID':
entities_to_add.append(self.add_solid(ent, scale_factor))
elif ent.dxftype() == 'POLYLINE':
entities_to_add.extend(self.add_polyline(ent, scale_factor))
elif ent.dxftype() == 'INSERT':
entities_to_add.extend(self.add_insertion(ent, doc.blocks, scale_factor))
elif ent.dxftype() == 'LWPOLYLINE':
entities_to_add.append(self.add_lwpolyline(ent, scale_factor))
# Adjust the view to fit the scene
self.preview_view.setTransform(QtGui.QTransform().scale(1, -1).rotate(0))
self.preview_scene.setSceneRect(self.preview_scene.itemsBoundingRect())
self.preview_view.fitInView(self.preview_scene.sceneRect(), Qt.KeepAspectRatio)
#self.preview_view.resetCachedContent()
self.preview_view.viewport().update()
#self.preview_view.update()
logger.debug("Successfully displayed DXF.")
except ezdxf.DXFError as dxf_error:
logger.error(f"DXF Error: {dxf_error}")
self.preview_scene.clear()
self.preview_scene.addText("Error reading DXF file")
#self.preview_scene.addText("Error reading DXF file")
except Exception as e:
logger.error(f"Unexpected Error: {type(e).__name__} - {e}")
self.preview_scene.clear()
self.preview_scene.addText("Unexpected error while displaying DXF file")
#self.preview_scene.addText("Unexpected error while displaying DXF file")
def calculate_scale_factor(self, msp, window_size):
#logger.info("Starting calculate_scale_factor diplay Dock// calculating scale factor")
min_x, min_y, max_x, max_y = float('inf'), float('inf'), float('-inf'), float('-inf')
for ent in msp:
if ent.dxftype() == 'LINE':
points = [ent.dxf.start, ent.dxf.end]
elif ent.dxftype() == 'CIRCLE' or ent.dxftype() == 'ARC':
center = ent.dxf.center
radius = ent.dxf.radius
points = [(center.x - radius, center.y - radius), (center.x + radius, center.y + radius)]
else:
# Add other entity types as needed
continue
for point in points:
min_x = min(min_x, point[0])
max_x = max(max_x, point[0])
min_y = min(min_y, point[1])
max_y = max(max_y, point[1])
if min_x == float('inf') or min_y == float('inf'):
return 1 # Return a default scale factor if no entities are found
scene_width = max_x - min_x
scene_height = max_y - min_y
scale_factor_x = window_size.width() / scene_width
scale_factor_y = window_size.height() / scene_height
#logger.info("End calculate_scale_factor display dock// calculating scale factor")
return min(scale_factor_x, scale_factor_y)
def add_line(self, line, scale_factor):
start_point = QPointF(line.dxf.start.x * scale_factor, line.dxf.start.y * scale_factor)
end_point = QPointF(line.dxf.end.x * scale_factor, line.dxf.end.y * scale_factor)
# Create a QGraphicsLineItem and set the pen width
line_item = self.preview_scene.addLine(start_point.x(), start_point.y(), end_point.x(), end_point.y())
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
line_item.setPen(pen)
return line_item
def add_circle(self, circle, scale_factor):
center = QPointF(circle.dxf.center.x * scale_factor, circle.dxf.center.y * scale_factor)
radius = circle.dxf.radius * scale_factor
# Create a QGraphicsEllipseItem and set the pen width
circle_item = self.preview_scene.addEllipse(center.x() - radius, center.y() - radius, 2 * radius, 2 * radius)
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
circle_item.setPen(pen)
return circle_item
def add_spline(self, spline, scale_factor):
# Convert control points to scaled coordinates
control_points = [(x * scale_factor, y * scale_factor) for x, y, _ in spline.control_points]
# Create a QPainterPath and add the spline
spline_path = QPainterPath()
spline_path.moveTo(*control_points[0])
for point in control_points[1:]:
spline_path.lineTo(*point)
# Create a QGraphicsPathItem and set the pen width
spline_item = QGraphicsPathItem(spline_path)
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
spline_item.setPen(pen)
# Add the spline item to the preview scene
self.preview_scene.addItem(spline_item)
return spline_item
def add_point(self, point, scale_factor):
point = QPointF(point.dxf.location.x * scale_factor, point.dxf.location.y * scale_factor)
return self.preview_scene.addEllipse(point.x() - 1, point.y() - 1, 2, 2)
def add_arc(self, arc, scale_factor):
center = QPointF(arc.dxf.center.x * scale_factor, arc.dxf.center.y * scale_factor)
radius = arc.dxf.radius * scale_factor
start_angle = arc.dxf.start_angle
end_angle = arc.dxf.end_angle
start_angle %= 360
end_angle %= 360
start_angle_rad = math.radians(start_angle)
end_angle_rad = math.radians(end_angle)
if start_angle > end_angle:
end_angle_rad += 2 * math.pi
arc_path = QPainterPath()
arc_path.arcMoveTo(center.x() - radius, center.y() - radius, 2 * radius, 2 * radius, -math.degrees(start_angle_rad))
arc_path.arcTo(center.x() - radius, center.y() - radius, 2 * radius, 2 * radius, -math.degrees(start_angle_rad), math.degrees(start_angle_rad - end_angle_rad))
# Create a QGraphicsPathItem and set the pen width
arc_item = QGraphicsPathItem(arc_path)
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
arc_item.setPen(pen)
# Add the arc item to the preview scene
self.preview_scene.addItem(arc_item)
return arc_item
def add_ellipse(self, ellipse, scale_factor):
# Centre de l'ellipse
center = QPointF(ellipse.dxf.center.x * scale_factor, ellipse.dxf.center.y * scale_factor)
# Vecteur de l'axe majeur
major_axis_vector = ellipse.dxf.major_axis # Vecteur non mis à l'échelle pour l'instant
major_axis_length = 2*math.hypot(major_axis_vector.x, major_axis_vector.y) * scale_factor
# Calcul de l'angle de rotation de l'ellipse
rotation_angle = math.degrees(math.atan2(major_axis_vector.y, major_axis_vector.x))
# Ratio entre l'axe mineur et l'axe majeur
ratio = ellipse.dxf.ratio
minor_axis_length = major_axis_length * ratio
# Création de l'ellipse centrée à l'origine
ellipse_item = QGraphicsEllipseItem(-major_axis_length / 2, -minor_axis_length / 2, major_axis_length, minor_axis_length)
# Positionnement de l'ellipse
ellipse_item.setPos(center)
# Application de la rotation
ellipse_item.setRotation(rotation_angle)
# Définition du stylo
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
ellipse_item.setPen(pen)
# Ajout de l'ellipse à la scène
self.preview_scene.addItem(ellipse_item)
return ellipse_item
def add_lwpolyline(self, lwpolyline, scale_factor):
points = []
bulge_info = []
for data in lwpolyline:
x, y, start_width, end_width, bulge = data
point = QPointF(x * scale_factor, y * scale_factor)
points.append(point)
bulge_info.append(bulge)
path = QPainterPath()
if points:
path.moveTo(points[0])
for i in range(1, len(points)):
bulge = bulge_info[i - 1]
if bulge != 0:
# Gestion des arcs via le bulge
# Calcul du segment d'arc entre points[i - 1] et points[i]
# Le bulge est la tangente de 1/4 de l'angle central de l'arc
# Pour une implémentation complète, il faudrait calculer les arcs ici
# Pour simplifier, nous traçons une ligne droite
path.lineTo(points[i])
else:
path.lineTo(points[i])
if lwpolyline.closed:
path.closeSubpath()
# Création de l'élément graphique
lwpolyline_item = QGraphicsPathItem(path)
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
lwpolyline_item.setPen(pen)
# Ajout de l'élément à la scène
self.preview_scene.addItem(lwpolyline_item)
return lwpolyline_item
def add_text(self, text_entity, scale_factor):
"""
Adds a text element to the preview scene using QGraphicsTextItem.
Args:
text_entity (Text): Text object containing the text and formatting information.
scale_factor (float): Scale factor to apply to the text.
Returns:
QGraphicsTextItem: The graphics item representing the text.
"""
insert_point = QPointF(text_entity.dxf.insert.x * scale_factor, text_entity.dxf.insert.y * scale_factor)
height = text_entity.dxf.height * scale_factor
rotation = text_entity.dxf.rotation
text_item = QGraphicsTextItem(text_entity.dxf.text)
text_item.setPos(insert_point)
font = QFont()
font.setPointSizeF(height)
text_item.setFont(font)
text_item.setRotation(rotation)
self.preview_scene.addItem(text_item)
return text_item
def add_mtext(self, mtext, scale_factor):
"""
Adds a multiline text element to the preview scene.
Args:
mtext (MText): MText object containing the multiline text and formatting information.
scale_factor (float): Scale factor to apply to the multiline text.
Returns:
QGraphicsItemGroup: The graphics item group representing the multiline text.
"""
insert_point = QPointF(mtext.dxf.insert.x * scale_factor, mtext.dxf.insert.y * scale_factor)
height = mtext.dxf.char_height * scale_factor
angle = math.degrees(mtext.dxf.rotation)
# Create a QGraphicsItemGroup to hold the multiline text
mtext_group = QGraphicsItemGroup()
# Create a QGraphicsTextItem for the multiline text and add it to the group
mtext_item = QGraphicsTextItem(mtext.plain_text())
mtext_item.setPos(insert_point.x(), insert_point.y())
mtext_item.setFont(QFont(mtext_item.font(), pointSize=height))
mtext_item.setRotation(angle)
mtext_group.addToGroup(mtext_item)
# Set the pen width for the multiline text
pen = QPen(Qt.black, 1, Qt.SolidLine)
pen.setCosmetic(True)
mtext_item.setPen(pen)
self.preview_scene.addItem(mtext_group)
return mtext_group
def add_polygon(self, points, scale_factor):
vertices = [QPointF(point.x * scale_factor, point.y * scale_factor) for point in points]
return self.preview_scene.addPolygon(QPolygonF(vertices))
def add_solid(self, solid, scale_factor):
vertices = [QPointF(vertex.x * scale_factor, vertex.y * scale_factor) for vertex in solid.get_points()]
return self.preview_scene.addPolygon(QPolygonF(vertices))
def add_insertion(self, insertion, doc, scale_factor):
block_name = insertion.dxf.name
block = doc.blocks.get(block_name)
items = []
if block:
for entity in block:
if entity.dxftype() == 'LINE':
items.append(self.add_line(entity, scale_factor))
elif entity.dxftype() == 'CIRCLE':
items.append(self.add_circle(entity, scale_factor))
return items
def add_polyline(self, polyline, scale_factor):
items = []
points = [vertex.dxf.location for vertex in polyline.vertices()]
for i in range(len(points) - 1):
start_point = QPointF(points[i].x * scale_factor, points[i].y * scale_factor)
end_point = QPointF(points[i + 1].x * scale_factor, points[i + 1].y * scale_factor)
line_item = self.preview_scene.addLine(start_point.x(), start_point.y(), end_point.x(), end_point.y())
items.append(line_item)
return items
def add_3dface(self, face, scale_factor):
vertices = [QPointF(vertex.x * scale_factor, vertex.y * scale_factor) for vertex in face.points()]
return self.preview_scene.addPolygon(QPolygonF(vertices))
def display_elements_in_preview(self, elements):
logger.info("Start display_elements_in_preview")
logger.debug("Clearing the preview scene.")
self.preview_scene.clear() # Clear the current scene
for element in elements:
self.preview_scene.addItem(element) # Add each element to the preview scene
logger.debug("Elements added to the preview scene.")
logger.info("End display_elements_in_preview")
def display_default_message(self):
logger.info("Start display_default_message")
logger.debug("Displaying a default message for unsupported files.")
self.preview_scene.clear()
self.preview_scene.addText("Preview not supported for this file type")
logger.info("End display_default_message")
def open_context_menu(self, position):
"""Ouvre un menu contextuel avec des options pour l'élément sélectionné.
Args:
position (QPoint): La position du clic droit de la souris.
"""
logger.info("Start open_context_menu")
try:
index = self.file_list.indexAt(position)
if not index.isValid():
logger.warning("Invalid index when opening the context menu")
return
menu = QMenu(self)
action_properties = menu.addAction("Properties")
openAction = QAction("Open", self)
openAction.triggered.connect(lambda: self.on_folder_double_clicked(index))
menu.addAction(openAction)
action_open_location = QAction("Open File Location", self)
action_open_location.triggered.connect(lambda: self.open_file_location(index))
menu.addAction(action_open_location)
action_open_location_on_windows = QAction("Open File Location on Windows", self)
action_open_location_on_windows.triggered.connect(lambda: self.open_file_location_on_windows(index))
menu.addAction(action_open_location_on_windows)
action = menu.exec_(self.file_list.viewport().mapToGlobal(position))
if action == action_properties:
logger.debug("Properties action selected")
# Obtenir le modèle actuel de la vue
model = self.file_list.model()
# Mapper l'index, toujours nécessaire car on est dans le contexte du menu
if isinstance(model, QSortFilterProxyModel) or isinstance(model, QAbstractProxyModel):
source_index = model.mapToSource(index)
else:
source_index = index
self.display_properties(source_index) # On passe l'index source
except Exception as e:
logger.error(f"Error opening context menu: {e}", exc_info=True)
logger.info("End open_context_menu")
def open_file_location_on_windows(self, index):
"""Ouvre l'emplacement du fichier dans l'explorateur de fichiers."""
logger.info("Start open_file_location_on_windows")
try:
model = self.file_list.model()
if isinstance(model, QSortFilterProxyModel) or isinstance(model, QAbstractProxyModel):
source_index = model.mapToSource(index)
else:
source_index = index
file_path = self.file_list_model.filePath(source_index)
folder_path = os.path.dirname(file_path)
logger.debug(f"Opening folder path: {folder_path}")
if os.path.exists(folder_path):
os.startfile(folder_path) # Ouvre l'emplacement dans l'explorateur
else:
logger.warning(f"Folder not found: {folder_path}")
QMessageBox.warning(self, "Error", "Folder not found.")
except Exception as e:
logger.error(f"Error opening file location: {e}", exc_info=True)
QMessageBox.critical(self, "Error", f"An error occurred: {str(e)}")
logger.info("End open_file_location_on_windows")
def open_file_location(self, index):
"""Ouvre l'emplacement du fichier dans la vue de la liste."""
try:
if self._in_search_mode:
logger.info("Sortie du mode recherche après open_file_location.")
self._in_search_mode = False
self.file_list.setModel(self.file_list_proxy)
self.search_box.clear()
if not index.isValid():
logger.error("L'index fourni est invalide.")
return
# Obtenir le chemin du fichier à partir de l'index
file_path = self.file_list_model.filePath(index)
logger.debug(f"Chemin du fichier obtenu : {file_path}")
# Extraire le chemin du dossier
folder_path = os.path.dirname(file_path)
logger.debug(f"Chemin du dossier extrait : {folder_path}")
# Obtenir l'index du dossier à partir du chemin
folder_index = self.file_list_model.index(folder_path)
logger.debug(f"Index du dossier obtenu : {folder_index}")
if folder_index.isValid():
# Ouvrir l'emplacement du fichier
self.file_list.setRootIndex(folder_index)
logger.info(f"Emplacement du fichier ouvert : {folder_path}")
else:
logger.error(f"Index de dossier invalide pour le chemin : {folder_path}")
except Exception as e:
logger.error(f"Erreur lors de l'ouverture de l'emplacement du fichier : {e}", exc_info=True)
@staticmethod
def readable_size(size):
logger.info("Function readable_size")
logger.debug("Converting size to readable format")
for unit in ['bytes', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB']:
if size < 1024.0:
return f"{size:.2f} {unit}"
size /= 1024.0
return f"{size:.2f} YB"
def display_properties(self, source_index): # On reçoit directement l'index source
logger.info("Start display_properties")
logger.debug("Starting to display file properties")
try:
# Vérifier si l'index source est valide
if not source_index.isValid():
logger.error("Invalid index provided.")
QMessageBox.warning(self, "Error", "Invalid selection.")
return
# Utiliser directement l'index source, plus besoin de mapper
file_path = self.file_list_model.filePath(source_index)
logger.debug(f"Attempting to display properties for: {file_path}")
file_info = QFileInfo(file_path)
if not file_info.exists():
logger.error(f"The file or folder does not exist: {file_path}")
QMessageBox.warning(self, "Error", "The file or folder does not exist.")
return
if not file_info.isReadable():
logger.error(f"Cannot access file or folder: {file_path}")
QMessageBox.warning(self, "Error", "Cannot access the file or folder.")
return
# Obtenir les propriétés de base
name = file_info.fileName()
if file_info.isDir():
size_str = "N/A"
file_type = "Folder"
else:
size_str = self.readable_size(file_info.size())
file_type = "File"
last_modified = file_info.lastModified().toString("dd/MM/yyyy HH:mm:ss")
# Construire le message à afficher
message = f"Type: {file_type}\nName: {name}\nSize: {size_str}\nLast Modified: {last_modified}"
# Afficher les propriétés dans une QMessageBox
QMessageBox.information(self, "Properties", message)
except Exception as e:
logger.error(f"Error displaying properties: {e}", exc_info=True)
QMessageBox.critical(self, "Error", f"An error occurred: {str(e)}")
logger.debug("Finished displaying properties")
logger.info("End display_properties")
# File system event handler class
class FileMonitor(FileSystemEventHandler):
def __init__(self, indexer, allowed_extensions = ALLOWED_EXTENSIONS):
super().__init__()
self.indexer = indexer
self.allowed_extensions = allowed_extensions
self.watched_dir = WATCHED_DIR
def on_any_event(self, event):
if not event.is_directory:
logger.info(f"Event detected on file: {event.src_path}, Event type: {event.event_type}")
if not self.indexer.is_indexing and not self.indexer.index_writer_lock.locked():
# Vérifier si le fichier modifié est dans WATCHED_DIR ou ses sous-répertoires
if os.path.commonpath([event.src_path, self.watched_dir]) == self.watched_dir:
#self.indexer.buildIndex()
logger.debug("FileMonitor.on_any_event(self, event) / self.indexer.buildIndex() not called")
else:
logger.warning("Indexation en cours, événement ignoré.")
def on_modified(self, event):
"""Met à jour l'index si un fichier pertinent est modifié."""
if not event.is_directory and os.path.commonpath([event.src_path, self.watched_dir]) == self.watched_dir and \
any(event.src_path.lower().endswith(ext) for ext in self.allowed_extensions):
logger.info(f"Modified file detected: {event.src_path}")
try:
logger.debug(f"Attempting to open index directory at {self.indexer.index_dir}")
ix = open_dir(self.indexer.index_dir)
with ix.searcher() as searcher:
with ix.writer() as writer:
# Lire le document existant pour récupérer les champs non modifiables
docnum = searcher.document_number(path=event.src_path) # Use searcher.document_number
logger.debug(f"Document number for {event.src_path}: {docnum}")
if docnum is not None:
logger.debug(f"Document found in index: {event.src_path}")
old_doc = searcher.stored_fields(docnum)
# Filter out fields not in the current schema
logger.debug(f"Old document fields before filtering: {old_doc}")
old_doc = {k: v for k, v in old_doc.items() if k in self.indexer.schema.names()}
logger.debug(f"Old document fields after filtering: {old_doc}")
# Update with new metadata
metadata = self.indexer.getMetadata(event.src_path)
if metadata:
logger.debug(f"Updating document in index: {event.src_path}, Metadata: {metadata}")
writer.update_document(path=event.src_path, **metadata) # Use the new metadata directly
logger.debug(f"File updated in index: {event.src_path}")
else:
logger.warning(f"Failed to get metadata for modified file: {event.src_path}")
except EmptyIndexError:
logger.warning(f"Index directory is empty, cannot update file: {event.src_path}")
except Exception as e:
logger.error(f"Error updating file in index: {e}")
def on_deleted(self, event):
"""Supprime un fichier de l'index s'il a été supprimé."""
if not event.is_directory and os.path.commonpath([event.src_path, self.watched_dir]) == self.watched_dir and \
any(event.src_path.lower().endswith(ext) for ext in self.allowed_extensions):
logger.info(f"Deleted file detected: {event.src_path}")
try:
ix = open_dir(self.indexer.index_dir)
with ix.writer() as writer:
writer.delete_by_term("path", event.src_path)
except EmptyIndexError:
logger.warning("Index directory is empty (cannot delete).")
except Exception as e:
logger.error(f"Error deleting file from index: {e}")
def on_created(self, event):
"""Ajoute un nouveau fichier à l'index si nécessaire."""
if not event.is_directory and os.path.commonpath([event.src_path, self.watched_dir]) == self.watched_dir and \
any(event.src_path.lower().endswith(ext) for ext in self.allowed_extensions):
logger.info(f"New file detected: {event.src_path}")
if self.indexer.is_indexing or self.indexer.index_writer_lock.locked():
# L'indexation est en cours, on met l'événement en attente
logger.warning(f"Indexing in progress, queuing creation event for {event.src_path}")
# Ici, vous pourriez ajouter le chemin du fichier à une file d'attente pour un traitement ultérieur
return # On sort de la fonction sans traiter l'événement pour le moment
try:
ix = open_dir(self.indexer.index_dir)
metadata = self.indexer.getMetadata(event.src_path)
if metadata:
with ix.writer() as writer:
# Conversion de la chaîne de temps de modification en objet datetime
modification_datetime = datetime.strptime(metadata["modification_time"], '%Y-%m-%d %H:%M:%S')
writer.add_document(
path=event.src_path,
name=os.path.basename(event.src_path).lower(),
ext=os.path.splitext(event.src_path)[1].lower(),
modification_time=modification_datetime, # Utilisation directe de l'objet datetime
creation_time=datetime.fromtimestamp(os.path.getctime(event.src_path)),
size=metadata['size'] # Assurez-vous que 'size' est présent dans les métadonnées
)
logger.debug(f"File added to index: {event.src_path}")
except EmptyIndexError:
logger.warning("Index directory is empty, creating index...")
self.indexer.buildIndex() # Reconstruire l'index si nécessaire
logger.debug("FileMonitor.on_created(self, event) / self.indexer.buildIndex() called")
except Exception as e:
logger.error(f"Error adding new file to index: {e}")
# File indexer class with metadata extraction
class FileIndexer(QObject):
indexUpdated = pyqtSignal()
indexerError = pyqtSignal(str)
indexingProgress = pyqtSignal(int, int) # Signal pour la progression (fichiers traités, total fichiers)
def __init__(self, parent=None, index_dir=INDEX_DIR, root_path=WATCHED_DIR):
super().__init__(parent)
self.index_dir = os.path.normpath(index_dir)
self.rootPath = os.path.normpath(root_path)
self.allowed_extensions = ALLOWED_EXTENSIONS
self.schema = Schema(
path=ID(stored=True, unique=True), # Chemin unique pour éviter les doublons
name=TEXT(stored=True), # Nom du fichier (pour la recherche textuelle)
ext=ID(stored=True), # Extension du fichier (pour le filtrage)
modification_time=DATETIME(stored=True), # Date de modification
creation_time=DATETIME(stored=True), # Date de création
size=STORED, # Ajouter la taille du fichier pour la comparaison
)
logger.debug(f"Schema defined with fields: {self.schema.names()}")
self.index_writer_lock = threading.Lock()
self.is_indexing = False
logger.debug(f"Initializing FileIndexer with index_dir: {self.index_dir}, rootPath: {self.rootPath}")
def buildIndex(self):
logger = logging.getLogger(__name__)
logger.debug("FileIndexer.buildIndex called")
if self.is_indexing or self.index_writer_lock.locked():
logger.warning("Indexation déjà en cours, appel à buildIndex ignoré.")
return
self.is_indexing = True
try:
with self.index_writer_lock:
start_time = time.time()
total_files_indexed = 0
total_files_updated = 0
ix = None
try:
# Open or create index
logger.debug(f"Checking if index exists at {self.index_dir} and if it is valid.")
if os.path.exists(self.index_dir) and exists_in(self.index_dir):
try:
ix = open_dir(self.index_dir)
logger.info("FileIndexer / Existing index found. Updating...")
except Exception as e:
logger.error(f"FileIndexer / Failed to open existing index: {e}", exc_info=True)
logger.info("FileIndexer / Corrupted index detected. Recreating index...")
self.recreate_index()
ix = create_in(self.index_dir, self.schema)
logger.info(f"FileIndexer / New index created in {self.index_dir}.")
else:
logger.info("FileIndexer / No index found. Creating a new index...")
self.clean_index_directory()
ix = create_in(self.index_dir, self.schema)
logger.info(f"FileIndexer / New index created in {self.index_dir}.")
assert ix is not None, "Index object should be initialized before proceeding."
# Index documents
with AsyncWriter(ix) as writer:
logger.debug("Writer successfully opened.")
allowed_files = []
for root, dirs, files in os.walk(self.rootPath):
#logger.debug(f"Scanning directory: {root}")
for filename in files:
full_path = os.path.normcase(os.path.normpath(os.path.join(root, filename)))
ext = os.path.splitext(full_path)[1].lower()
if ext in self.allowed_extensions:
allowed_files.append(full_path)
#logger.debug(f"Found file: {full_path}, extension: {ext}")
total_files = len(allowed_files)
processed_files = 0
for full_path in allowed_files:
ext = os.path.splitext(full_path)[1].lower()
try:
metadata = self.getMetadata(full_path)
if not metadata:
logger.warning(f"FileIndexer / Failed to get metadata for: {full_path}")
continue
try:
modification_datetime = datetime.strptime(metadata["modification"], '%Y-%m-%d %H:%M:%S')
creation_datetime = datetime.fromtimestamp(os.path.getctime(full_path))
except (ValueError, KeyError) as date_error:
logger.error(f"FileIndexer / Error processing date for {full_path}: {date_error}")
continue
# Check if document exists
with ix.searcher() as searcher:
doc_exists = searcher.document(path=os.path.normcase(os.path.normpath(full_path))) is not None
#logger.debug(f"Indexing document: path={full_path}, name={os.path.basename(full_path).lower()}, ext={ext}, "
#f"modification_time={modification_datetime}, creation_time={creation_datetime}, size={metadata['size']}")
if doc_exists:
# Update existing document
writer.update_document(
path=os.path.normcase(os.path.normpath(full_path)),
name=os.path.basename(full_path).lower(),
ext=ext,
modification_time=modification_datetime,
creation_time=creation_datetime,
size=metadata['size']
)
total_files_updated += 1
#logger.info(f"Updated document: {full_path}")
else:
# Add a new document
writer.add_document(
path=os.path.normcase(os.path.normpath(full_path)),
name=os.path.basename(full_path).lower(),
ext=ext,
modification_time=modification_datetime,
creation_time=creation_datetime,
size=metadata['size']
)
total_files_indexed += 1
#logger.info(f"Added document: {full_path}")
# Log the size of the index
#current_doc_count = ix.doc_count()
#logger.debug(f"Current index size (number of documents): {current_doc_count}")
processed_files += 1
if processed_files % 100 == 0 and processed_files > 0:
logger.info(f"FileIndexer / Indexed {processed_files} out of {total_files} files...")
QMetaObject.invokeMethod(self, "indexingProgress", Qt.QueuedConnection,
Q_ARG(int, processed_files), Q_ARG(int, total_files))
except Exception as process_error:
logger.error(f"FileIndexer / Error processing {full_path}: {process_error}", exc_info=True)
continue
except Exception as indexing_error:
logger.error(f"FileIndexer / An error occurred during index setup or writing: {indexing_error}", exc_info=True)
raise
end_time = time.time()
logger.info(f"File indexing completed in {end_time - start_time:.2f} seconds.")
logger.info(f"FileIndexer / Total files indexed: {total_files_indexed}, updated: {total_files_updated}")
self.indexUpdated.emit()
except Exception as e:
logger.error(f"FileIndexer / An error occurred during indexing: {e}")
self.indexerError.emit(str(e))
finally:
self.is_indexing = False
logger.debug("FileIndexer / Indexing finished, resetting is_indexing flag.")
def recreate_index(self):
"""Helper method to recreate the index directory."""
try:
if os.path.exists(self.index_dir):
shutil.rmtree(self.index_dir)
os.makedirs(self.index_dir, exist_ok=True)
except Exception as e:
logger.error(f"Failed to recreate index directory: {e}")
raise
def clean_index_directory(self):
"""Clean index directory if it exists."""
if os.path.exists(self.index_dir):
logger.warning("Index directory exists but no index found. Cleaning up the directory.")
for filename in os.listdir(self.index_dir):
file_path = os.path.join(self.index_dir, filename)
try:
if os.path.isfile(file_path) or os.path.islink(file_path):
os.unlink(file_path)
elif os.path.isdir(file_path):
shutil.rmtree(file_path)
except Exception as e:
logger.error(f'Failed to delete {file_path}. Reason: {e}')
def getMetadata(self, filePath):
try:
if not os.path.exists(filePath):
logger.warning(f"File {filePath} does not exist.")
return None
stat = os.stat(filePath)
size = stat.st_size # Size in bytes
modificationTime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(stat.st_mtime))
metadata = {'size': size, 'modification': modificationTime}
#logger.debug(f"Metadata obtained for {filePath}: {metadata}")
return metadata
except OSError as e:
logger.error(f"Error accessing file {filePath}: {e}")
return None
def search(self, criteria, criteria_type="filename"):
logger.info(f"FileIndexer/ Starting search with criteria: {criteria}, criteria_type: {criteria_type}")
results = []
# Attendre que l'indexation en cours se termine
while self.is_indexing:
logger.info("Indexation en cours, en attente de sa complétion avant de lancer la recherche...")
time.sleep(1) # Attendre 1 seconde avant de vérifier à nouveau
try:
if not exists_in(self.index_dir):
logger.warning("Index non trouvé. Lancement de l'indexation...")
self.buildIndex()
# Attendre la fin de l'indexation
while self.is_indexing:
logger.info("Indexation en cours, en attente de sa complétion...")
time.sleep(1)
ix = open_dir(self.index_dir)
except EmptyIndexError as e:
logger.error(f"Erreur lors de l'ouverture de l'index: {e}", exc_info=True)
return results
with ix.searcher() as searcher:
query_parts = []
criteria_lower = self.criteria.lower().strip()
if criteria_type == "filename":
query_part = FuzzyTerm("name", criteria_lower, maxdist=1)
query_parts.append(query_part)
elif criteria_type in ("creation_date", "modification_date"):
date_field = "creation_time" if criteria_type == "creation_date" else "modification_time"
query_parts.append(DateRange(date_field, self.start_date, self.end_date))
# Filtrage par extension
extension_terms = [Term("ext", ext.lower()) for ext in self.allowed_extensions]
query_parts.append(Or(extension_terms))
# Ajout du filtrage par chemin si nécessaire
if self.rootPath:
path_query = Term("path", self.rootPath)
query_parts.append(path_query)
final_query = And(query_parts)
logger.debug(f"Final Whoosh query: {final_query}")
hits = searcher.search(final_query, limit=100)
results = [hit["path"] for hit in hits] # Récupération des résultats
return results
@pyqtSlot()
def buildIndexAsync(self):
logger.debug("FileIndexer.buildIndexAsync called")
if self.is_indexing:
logger.warning("Indexation déjà en cours, appel à buildIndexAsync ignoré.")
return
logger.info("FileIndexer / Starting asynchronous indexing in a separate thread...")
try:
self.indexThread = QThread(self)
worker = IndexWorker(self)
worker.moveToThread(self.indexThread)
# Connexion des signaux
self.indexThread.started.connect(worker.run)
worker.indexingFinished.connect(self.indexThread.quit)
worker.indexingFinished.connect(worker.deleteLater)
self.indexThread.finished.connect(self.indexThread.deleteLater)
worker.indexingFinished.connect(lambda: self.indexUpdated.emit())
worker.error.connect(self.indexerError.emit)
worker.indexingProgress.connect(self.indexingProgress.emit)
worker.indexingFinished.connect(lambda: (setattr(self, 'is_indexing', False), logger.debug("FileIndexer / Indexing finished, resetting is_indexing flag.")))
self.indexThread.start()
logger.debug(f"FileIndexer / Starting indexThread.")
except Exception as e:
logger.error(f"FileIndexer / Error starting IndexThread: {e}")
def inspect_index(self):
"""Inspecte tous les documents dans l'index et affiche leurs champs stockés."""
logger.debug(f"FileIndexer / inspect_index.")
try:
ix = open_dir(self.index_dir)
with ix.searcher() as searcher:
for docnum in range(searcher.doc_count()):
stored_fields = searcher.stored_fields(docnum)
#logger.debug(f"Document {docnum}: {stored_fields}")
except Exception as e:
logger.error(f"Error inspecting index: {e}")
class IndexWorker(QObject): # classe pour le worker d'indexation
"""Worker pour l'indexation asynchrone des fichiers."""
indexingFinished = pyqtSignal()
error = pyqtSignal(str)
indexingProgress = pyqtSignal(int, int)
def __init__(self, indexer):
super().__init__()
self.indexer = indexer
#self.rootPath = rootPath
@pyqtSlot()
def run(self):
logger.debug(f"IndexWorker.run(self) / run method started.")
logger.info("IndexWorker.run(self) / Indexing started in the background thread.")
try:
logger.debug("IndexWorker.run(self) / Calling buildIndex...")
logger.debug("IndexWorker.run(self) / About to call self.indexer.buildIndex")
self.indexer.buildIndex() # Appeler la méthode buildIndex dans le thread
logger.debug("IndexWorker.run(self) / self.indexer.buildIndex call completed")
logger.debug("IndexWorker.run(self) / buildIndex completed.")
except Exception as e:
logger.error(f"IndexWorker.run(self) / Error during indexing: {e}", exc_info=True) # Log the full traceback
self.error.emit(str(e))
finally:
self.indexingFinished.emit()
logger.debug("IndexWorker.run(self) / Emitting indexingFinished signal.")
logger.info("IndexWorker / run method finished.")
class MeasureMode(Enum):
NONE = 0
DISTANCE = 1
ANGLE = 2
AREA = 3
class CustomGraphicsView(QGraphicsView):
resized = pyqtSignal()
def __init__(self, parent=None, zoom_in_factor=1.15, unit_scale=1.0, unit_name="unités"):
super().__init__(parent)
self.setDragMode(QGraphicsView.ScrollHandDrag)
self.setRenderHints(QPainter.Antialiasing | QPainter.SmoothPixmapTransform)
self.setTransformationAnchor(QGraphicsView.AnchorUnderMouse)
self.setResizeAnchor(QGraphicsView.AnchorUnderMouse)
self.zoom_in_factor = zoom_in_factor
self.zoom_out_factor = 1 / zoom_in_factor
self._initial_view_fitted = False
self.start_point = None
self.temp_line = None
self.measurement_label = QLabel(self)
self.measurement_label.setStyleSheet("background-color: rgba(255, 255, 255, 200);")
self.measurement_label.hide()
self.setMouseTracking(True)
self.selected_line1 = None # Pour la mesure d'angle
self.spatial_index = {}
self.previously_highlighted = None
self._adjustment_in_progress = False
self._signal_connected = False
self.measure_mode = MeasureMode.NONE
self.unit_scale = unit_scale
self.unit_name = unit_name
self.measure_items = [] # Liste pour stocker les éléments de mesure
self.original_pens = {}
def set_measure_mode(self, mode):
self.measure_mode = mode
self.measurement_label.hide() # Cacher l'étiquette lors du changement de mode
def is_circle(self, item):
"""Vérifie si un QGraphicsEllipseItem est un cercle."""
try:
rect = item.rect()
return rect.width() == rect.height()
except Exception as e:
logger.error(f"Error in is_circle: {e}", exc_info=True)
return False
def measure_circle_radius(self, circle_item, scene_pos):
"""Mesure le rayon du cercle et affiche la mesure."""
try:
rect = circle_item.rect()
center = rect.center()
radius = rect.width() / 2
real_radius = radius * self.unit_scale
self.display_radius(real_radius, scene_pos)
except Exception as e:
logger.error(f"Error in measure_circle_radius: {e}", exc_info=True)
def display_radius(self, radius, position):
"""Affiche le rayon à la position spécifiée."""
try:
self.measurement_label.setText(f"Rayon: {radius:.2f} {self.unit_name}")
self.measurement_label.adjustSize()
# Convertir la position de la scène en position de la vue
screen_position = self.mapFromScene(position)
# Assurez-vous que l'étiquette reste dans les limites de la vue
view_rect = self.viewport().rect()
label_width = self.measurement_label.width()
label_height = self.measurement_label.height()
# Ajustez la position si l'étiquette dépasse les limites de la vue
if screen_position.x() + label_width > view_rect.right():
screen_position.setX(view_rect.right() - label_width)
if screen_position.y() + label_height > view_rect.bottom():
screen_position.setY(view_rect.bottom() - label_height)
if screen_position.x() < view_rect.left():
screen_position.setX(view_rect.left())
if screen_position.y() < view_rect.top():
screen_position.setY(view_rect.top())
# Déplacer l'étiquette à la position ajustée
adjusted_position = self.adjust_label_position(screen_position)
self.measurement_label.move(adjusted_position)
self.measurement_label.show()
except Exception as e:
logger.error(f"Error in display_radius: {e}", exc_info=True)
def display_measurement(self, text, item):
"""Affiche la mesure à côté de l'élément."""
try:
bounding_rect = item.boundingRect()
center = bounding_rect.center()
screen_position = self.mapFromScene(center)
self.measurement_label.setText(text)
self.measurement_label.adjustSize()
self.measurement_label.move(screen_position)
self.measurement_label.show()
except Exception as e:
logger.error(f"Error in display_measurement: {e}", exc_info=True)
def add_control_point(self, position, connected_line, is_start_point):
"""Ajoute un point de contrôle interactif à la position donnée."""
try:
logger.debug(f"Adding control point at {position} to line {connected_line}")
control_point = ControlPoint(position, connected_line)
if self.scene():
self.scene().addItem(control_point)
if is_start_point:
connected_line.start_control_point = control_point
logger.debug("Assigned as start_control_point")
else:
connected_line.end_control_point = control_point
logger.debug("Assigned as end_control_point")
except Exception as e:
logger.error(f"Error in add_control_point: {e}", exc_info=True)
def add_arrow(self, start, end):
"""Ajoute une flèche à chaque extrémité d'une ligne."""
try:
line_vector = end - start
length = math.sqrt(line_vector.x() ** 2 + line_vector.y() ** 2)
if length == 0:
return
# Normaliser le vecteur de ligne
line_vector /= length
# Définir les points de flèche
arrow_size = 10
perp_vector = QPointF(-line_vector.y(), line_vector.x()) * arrow_size
# Flèche au début (pointant vers l'intérieur)
arrow_p1 = start + line_vector * arrow_size + perp_vector
arrow_p2 = start + line_vector * arrow_size - perp_vector
arrow_head = QPolygonF([start, arrow_p1, arrow_p2])
arrow_item = QGraphicsPolygonItem(arrow_head)
arrow_item.setBrush(QBrush(QColor("blue")))
self.scene().addItem(arrow_item)
# Flèche à la fin (pointant vers l'extérieur)
arrow_p1 = end - line_vector * arrow_size + perp_vector
arrow_p2 = end - line_vector * arrow_size - perp_vector
arrow_head = QPolygonF([end, arrow_p1, arrow_p2])
arrow_item = QGraphicsPolygonItem(arrow_head)
arrow_item.setBrush(QBrush(QColor("blue")))
self.scene().addItem(arrow_item)
except Exception as e:
logger.error(f"Error in add_arrow: {e}", exc_info=True)
def mousePressEvent(self, event):
try:
if self.scene() is None:
return
scene_pos = self.mapToScene(event.pos())
closest_entity, closest_point = self.find_closest_entity(scene_pos)
if event.button() == Qt.LeftButton:
if self.measure_mode == MeasureMode.DISTANCE:
self.start_point = closest_point if closest_entity else scene_pos
self.measurement_label.hide()
self.temp_line = QGraphicsLineItem(QLineF(self.start_point, self.start_point))
self.temp_line.setPen(QPen(QColor("blue"), 2, Qt.DashLine))
self.scene().addItem(self.temp_line)
elif self.measure_mode == MeasureMode.AREA and isinstance(closest_entity, QGraphicsPolygonItem):
self.measure_polygon(closest_entity)
elif event.button() == Qt.RightButton and isinstance(closest_entity, QGraphicsLineItem):
self.selected_line1 = closest_entity
except Exception as e:
logger.error(f"Error in mousePressEvent: {e}", exc_info=True)
super().mousePressEvent(event)
def mouseMoveEvent(self, event):
try:
scene_pos = self.mapToScene(event.pos())
closest_entity, _ = self.find_closest_entity(scene_pos)
if self.previously_highlighted:
if self.scene() and self.previously_highlighted in self.scene().items():
self.reset_item_appearance(self.previously_highlighted)
self.previously_highlighted = None
if closest_entity:
self.highlight_item(closest_entity)
self.previously_highlighted = closest_entity
if self.measure_mode == MeasureMode.DISTANCE and self.start_point and self.temp_line:
current_point = self.mapToScene(event.pos())
snap_point = self.find_closest_snap_point(current_point)
self.temp_line.setLine(QLineF(self.start_point, snap_point))
dx = snap_point.x() - self.start_point.x()
dy = snap_point.y() - self.start_point.y()
distance = math.sqrt(dx * dx + dy * dy)
self.update_measurement_label(distance, self.start_point, snap_point)
super().mouseMoveEvent(event)
except Exception as e:
logger.error(f"Error in mouseMoveEvent: {e}", exc_info=True)
def mouseReleaseEvent(self, event):
try:
if self.measure_mode == MeasureMode.DISTANCE and self.start_point and self.temp_line:
end_point = self.mapToScene(event.pos())
closest_entity, closest_point = self.find_closest_entity(end_point)
end_point = closest_point if closest_entity else end_point
self.scene().removeItem(self.temp_line)
final_line = MeasuredLine(QLineF(self.start_point, end_point))
final_line.setPen(QPen(QColor("blue"), 2))
self.scene().addItem(final_line)
self.add_arrow(self.start_point, end_point)
logger.debug(f"Avant add_control_point (start): {final_line}") # Vérifier l'état de final_line
self.add_control_point(self.start_point, final_line, is_start_point=True)
logger.debug(f"Après add_control_point (start): {final_line}") # Vérifier l'état de final_line
logger.debug(f"Avant add_control_point (end): {final_line}") # Vérifier l'état de final_line
self.add_control_point(end_point, final_line, is_start_point=False)
logger.debug(f"Après add_control_point (end): {final_line}") # Vérifier l'état de final_line
dx = end_point.x() - self.start_point.x()
dy = end_point.y() - self.start_point.y()
distance = math.sqrt(dx * dx + dy * dy)
self.update_measurement_label(distance, self.start_point, end_point)
# Ajouter les éléments à la liste de mesure
self.measure_items.extend([final_line, final_line.start_control_point, final_line.end_control_point])
self.start_point = None
self.temp_line = None
elif event.button() == Qt.RightButton and self.selected_line1:
scene_pos = self.mapToScene(event.pos())
closest_line, _ = self.find_closest_entity(scene_pos)
if closest_line and closest_line != self.selected_line1:
angle = self.calculate_angle_between_lines(self.selected_line1.line(), closest_line.line())
self.display_angle(angle, event.pos())
self.selected_line1 = None
super().mouseReleaseEvent(event)
except Exception as e:
logger.error(f"Error in mouseReleaseEvent: {e}", exc_info=True)
def clear_measurements(self):
"""Supprime tous les éléments de mesure de la scène."""
try:
logger.debug("Clearing all measurements")
for item in list(self.measure_items):
if self.scene() and self.scene().contains(item):# Vérifier si la scène existe avant de supprimer l'élément
self.scene().removeItem(item)
self.measure_items.clear()
self.measurement_label.hide()
except Exception as e:
logger.error(f"Error in clear_measurements: {e}", exc_info=True)
def highlight_item(self, item):
try:
if item not in self.original_pens:
self.original_pens[item] = item.pen()
item.setPen(QPen(QColor("orange"), 2))
except Exception as e:
logger.error(f"Error in highlight_item: {e}", exc_info=True)
def reset_item_appearance(self, item):
try:
if item in self.original_pens:
item.setPen(self.original_pens.pop(item))
except Exception as e:
logger.error(f"Error in reset_item_appearance: {e}", exc_info=True)
def build_spatial_index(self):
try:
self.spatial_index = {}
if self.scene():
for item in self.scene().items():
bounding_box = item.mapToScene(item.boundingRect()).boundingRect()
key = (int(math.floor(bounding_box.x() / 100)), int(math.floor(bounding_box.y() / 100)))
self.spatial_index.setdefault(key, []).append(item)
except Exception as e:
logger.error(f"Error in build_spatial_index: {e}", exc_info=True)
def find_items_in_area(self, area):
try:
# Calculer les clés pour les coins supérieur gauche et inférieur droit
top_left_key = (int(area.left() // 100), int(area.top() // 100))
bottom_right_key = (int(area.right() // 100), int(area.bottom() // 100))
# Assurer que les plages de clés sont correctes
x_start = min(top_left_key[0], bottom_right_key[0])
x_end = max(top_left_key[0], bottom_right_key[0])
y_start = min(top_left_key[1], bottom_right_key[1])
y_end = max(top_left_key[1], bottom_right_key[1])
# Collecter toutes les clés dans la plage
keys = []
for x in range(x_start, x_end + 1):
for y in range(y_start, y_end + 1):
keys.append((x, y))
# Récupérer les éléments pour chaque clé
items = []
for key in keys:
items.extend(self.spatial_index.get(key, []))
# Filtrer les éléments qui intersectent réellement la zone
return [item for item in items if item.sceneBoundingRect().intersects(area)]
except Exception as e:
logger.error(f"Error in find_items_in_area: {e}", exc_info=True)
return []
def find_closest_entity(self, scene_pos, search_radius=500):
closest_entity = None
closest_point = None
min_distance = float('inf')
try:
search_area = QRectF(scene_pos.x() - search_radius, scene_pos.y() - search_radius,
2 * search_radius, 2 * search_radius)
items_in_area = self.find_items_in_area(search_area)
for item in items_in_area:
if isinstance(item, QGraphicsLineItem):
line = item.line()
point_on_line = self.closest_point_on_line(line, scene_pos)
distance = self.euclidean_distance(point_on_line, scene_pos)
if distance < min_distance:
min_distance = distance
closest_entity = item
closest_point = point_on_line
elif isinstance(item, QGraphicsEllipseItem):
ellipse = item.rect()
if ellipse.width() == ellipse.height():
center = ellipse.center()
radius = ellipse.width() / 2
point_on_circle = self.closest_point_on_circle(center, radius, scene_pos)
distance = self.euclidean_distance(point_on_circle, scene_pos)
if distance < min_distance:
min_distance = distance
closest_entity = item
closest_point = point_on_circle
else:
center = ellipse.center()
distance = self.euclidean_distance(center, scene_pos)
if distance < min_distance:
min_distance = distance
closest_entity = item
closest_point = center
elif isinstance(item, QGraphicsPolygonItem):
polygon = item.polygon()
# Utilisez un point moyen ou une méthode plus précise
center = polygon.boundingRect().center()
distance = self.euclidean_distance(center, scene_pos)
if distance < min_distance:
min_distance = distance
closest_entity = item
closest_point = center
elif isinstance(item, QGraphicsRectItem):
rect = item.rect()
point_on_rect = self.closest_point_on_rect(rect, scene_pos)
distance = self.euclidean_distance(point_on_rect, scene_pos)
if distance < min_distance:
min_distance = distance
closest_entity = item
closest_point = point_on_rect
elif isinstance(item, QGraphicsPathItem):
path = item.path()
point_on_path = self.closest_point_on_path(path, scene_pos)
if point_on_path is not None:
distance = self.euclidean_distance(point_on_path, scene_pos)
if distance < min_distance:
min_distance = distance
closest_entity = item
closest_point = point_on_path
# Ajoutez d'autres types d'entités et leurs calculs ici
else:
continue
except Exception as e:
logger.error(f"Error in find_closest_entity: {e}", exc_info=True)
return closest_entity, closest_point
def find_closest_snap_point(self, scene_pos, search_radius=10):
closest_point = None
min_distance = float('inf')
try:
search_area = QRectF(scene_pos.x() - search_radius, scene_pos.y() - search_radius,
2 * search_radius, 2 * search_radius)
for item in self.scene().items(search_area):
if isinstance(item, QGraphicsLineItem):
for point in [item.line().p1(), item.line().p2()]:
distance = self.euclidean_distance(point, scene_pos)
if distance < min_distance:
min_distance = distance
closest_point = point
elif isinstance(item, QGraphicsEllipseItem):
center = item.rect().center()
distance = self.euclidean_distance(center, scene_pos)
if distance < min_distance:
min_distance = distance
closest_point = center
# Ajouter d'autres types d'entités si nécessaire
except Exception as e:
logger.error(f"Error in find_closest_snap_point: {e}", exc_info=True)
return closest_point if min_distance <= search_radius else scene_pos
def closest_point_on_line(self, line, point):
try:
line_vector = line.p2() - line.p1()
point_vector = point - line.p1()
line_length_squared = line_vector.x() ** 2 + line_vector.y() ** 2
if line_length_squared == 0:
return line.p1()
t = QPointF.dotProduct(point_vector, line_vector) / line_length_squared
t = max(0, min(1, t))
closest = line.p1() + line_vector * t
return closest
except Exception as e:
logger.error(f"Error in closest_point_on_line: {e}", exc_info=True)
return point # Retourner le point original en cas d'erreur
def closest_point_on_circle(self, center, radius, point):
try:
direction = point - center
distance = math.sqrt(direction.x() ** 2 + direction.y() ** 2)
if distance == 0:
return center
return center + direction * (radius / distance)
except Exception as e:
logger.error(f"Error in closest_point_on_circle: {e}", exc_info=True)
return point
def closest_point_on_rect(self, rect, point):
try:
left = rect.left()
right = rect.right()
top = rect.top()
bottom = rect.bottom()
x = min(max(point.x(), left), right)
y = min(max(point.y(), top), bottom)
return QPointF(x, y)
except Exception as e:
logger.error(f"Error in closest_point_on_rect: {e}", exc_info=True)
return point
def closest_point_on_path(self, path, point):
try:
if path.isEmpty():
return None
min_distance_sq = float('inf')
closest_point = None
last_point = None
i = 0
element_count = path.elementCount()
while i < element_count:
element = path.elementAt(i)
if element.type == QPainterPath.ElementType.CurveToElement:
if i + 2 < element_count and path.elementAt(i + 2).type == QPainterPath.ElementType.CurveToDataElement:
# Courbe cubique
element2 = path.elementAt(i + 1)
element3 = path.elementAt(i + 2)
cp1 = last_point
cp2 = QPointF(element.x, element.y)
cp3 = QPointF(element2.x, element2.y)
end_point = QPointF(element3.x, element3.y)
p_on_curve = self.closest_point_on_bezier_curve(cp1, cp2, cp3, end_point, point)
dist_sq = (p_on_curve.x() - point.x()) ** 2 + (p_on_curve.y() - point.y()) ** 2
if dist_sq < min_distance_sq:
min_distance_sq = dist_sq
closest_point = p_on_curve
i += 3
else:
# Courbe quadratique
if i + 1 < element_count:
element2 = path.elementAt(i + 1)
cp1 = last_point
cp2 = QPointF(element.x, element.y)
end_point = QPointF(element2.x, element2.y)
p_on_curve = self.closest_point_on_bezier_curve(cp1, cp2, end_point, point, segments=100)
dist_sq = (p_on_curve.x() - point.x()) ** 2 + (p_on_curve.y() - point.y()) ** 2
if dist_sq < min_distance_sq:
min_distance_sq = dist_sq
closest_point = p_on_curve
i += 2
else:
i += 1
elif element.type == QPainterPath.ElementType.MoveToElement:
last_point = QPointF(element.x, element.y)
i += 1
else:
if last_point is not None:
line = QLineF(last_point, QPointF(element.x, element.y))
p_on_line = self.closest_point_on_line(line, point)
dist_sq = (p_on_line.x() - point.x()) ** 2 + (p_on_line.y() - point.y()) ** 2
if dist_sq < min_distance_sq:
min_distance_sq = dist_sq
closest_point = p_on_line
last_point = QPointF(element.x, element.y)
i += 1
return closest_point
except Exception as e:
logger.error(f"Error in closest_point_on_path: {e}", exc_info=True)
return point
def closest_point_on_bezier_curve(self, cp1, cp2, cp3, end_point, point, segments=100):
min_distance_sq = float('inf')
closest_point = None
try:
for i in range(segments + 1):
t = i / segments
x = (1 - t)**3 * cp1.x() + 3 * (1 - t)**2 * t * cp2.x() + 3 * (1 - t) * t**2 * cp3.x() + t**3 * end_point.x()
y = (1 - t)**3 * cp1.y() + 3 * (1 - t)**2 * t * cp2.y() + 3 * (1 - t) * t**2 * cp3.y() + t**3 * end_point.y()
p_on_curve = QPointF(x, y)
dist_sq = (p_on_curve.x() - point.x())**2 + (p_on_curve.y() - point.y())**2
if dist_sq < min_distance_sq:
min_distance_sq = dist_sq
closest_point = p_on_curve
return closest_point
except Exception as e:
logger.error(f"Error in closest_point_on_bezier_curve: {e}", exc_info=True)
return point
def euclidean_distance(self, p1, p2):
"""Calcule la distance euclidienne entre deux points."""
try:
return math.sqrt((p1.x() - p2.x()) ** 2 + (p1.y() - p2.y()) ** 2)
except Exception as e:
logger.error(f"Error in euclidean_distance: {e}", exc_info=True)
return 0
def calculate_angle_between_lines(self, line1, line2):
try:
vector1 = QPointF(line1.dx(), line1.dy())
vector2 = QPointF(line2.dx(), line2.dy())
dot_product = QPointF.dotProduct(vector1, vector2)
length1 = math.sqrt(QPointF.dotProduct(vector1, vector1))
length2 = math.sqrt(QPointF.dotProduct(vector2, vector2))
if length1 == 0 or length2 == 0:
return 0
cos_theta = dot_product / (length1 * length2)
# Clamp cos_theta to the valid range to avoid math domain error
cos_theta = max(-1.0, min(1.0, cos_theta))
angle = math.acos(cos_theta)
return angle
except Exception as e:
logger.error(f"Error in calculate_angle_between_lines: {e}", exc_info=True)
return 0
def measure_polygon(self, polygon_item):
try:
polygon = polygon_item.polygon()
perimeter = 0
area = 0
n = polygon.count()
for i in range(n):
p1 = polygon.at(i)
p2 = polygon.at((i + 1) % n)
perimeter += self.euclidean_distance(p1, p2)
area += (p1.x() * p2.y()) - (p2.x() * p1.y())
area = abs(area) / 2
real_perimeter = perimeter * self.unit_scale
real_area = area * (self.unit_scale ** 2)
self.display_measurement(
f"Périmètre: {real_perimeter:.2f} {self.unit_name}, "
f"Aire: {real_area:.2f} {self.unit_name}²",
polygon_item
)
except Exception as e:
logger.error(f"Error in measure_polygon: {e}", exc_info=True)
def adjust_label_position(self, screen_position):
try:
view_rect = self.viewport().rect()
label_width = self.measurement_label.width()
label_height = self.measurement_label.height()
x = screen_position.x()
y = screen_position.y()
# Vérifier si l'étiquette dépasse les bords droit et inférieur de la vue
if x + label_width > view_rect.right():
x = view_rect.right() - label_width
if y + label_height > view_rect.bottom():
y = view_rect.bottom() - label_height
# Vérifier si l'étiquette dépasse les bords gauche et supérieur de la vue
if x < view_rect.left():
x = view_rect.left()
if y < view_rect.top():
y = view_rect.top()
return QPoint(int(x), int(y))
except Exception as e:
logger.error(f"Error in adjust_label_position: {e}", exc_info=True)
return screen_position # Retourner la position originale en cas d'erreur
def display_angle(self, angle, position):
try:
angle_degrees = math.degrees(angle)
self.measurement_label.setText(f"Angle: {angle_degrees:.2f}°")
self.measurement_label.adjustSize()
adjusted_position = self.adjust_label_position(position)
self.measurement_label.move(adjusted_position)
self.measurement_label.show()
except Exception as e:
logger.error(f"Error in display_angle: {e}", exc_info=True)
def update_measurement_label(self, distance, start_point, end_point):
try:
# Obtenez le facteur d'échelle depuis l'instance de FileViewer
# Assurez-vous que le parent de ce CustomGraphicsView est bien FileViewer
file_viewer = self.window() # Utiliser self.window() pour accéder à la fenêtre principale
if hasattr(file_viewer, 'scale_factor'):
scale_factor = file_viewer.scale_factor
real_distance = distance / scale_factor # Appliquer le facteur d'échelle
else:
real_distance = distance # Si le facteur d'échelle n'est pas disponible, utilisez la distance brute
# real_distance = distance * self.unit_scale
self.measurement_label.setText(f"Distance: {real_distance:.2f} {self.unit_name}")
self.measurement_label.adjustSize()
mid_x = (start_point.x() + end_point.x()) / 2
mid_y = (start_point.y() + end_point.y()) / 2
screen_position = self.mapFromScene(QPointF(mid_x, mid_y))
label_position = QPoint(int(screen_position.x() - self.measurement_label.width() / 2),
int(screen_position.y() - self.measurement_label.height() / 2))
adjusted_position = self.adjust_label_position(label_position)
self.measurement_label.move(adjusted_position)
self.measurement_label.show()
except Exception as e:
logger.error(f"Error in update_measurement_label: {e}", exc_info=True)
def wheelEvent(self, event):
try:
# Définir les limites de zoom
zoomOutScale = 0.07 # Échelle minimale autorisée
zoomInScale = 100 # Échelle maximale autorisée
# Obtenir l'échelle actuelle
old_scale = self.transform().m11()
oldPos = self.mapToScene(event.pos()) # Position avant le zoom
if event.angleDelta().y() > 0 and old_scale < zoomInScale:
factor = self.zoom_in_factor
elif event.angleDelta().y() < 0 and old_scale > zoomOutScale:
factor = self.zoom_out_factor
else:
factor = 1 # Pas de zoom si les limites sont atteintes
self.scale(factor, factor)
# Recentrer la vue pour que le point sous la souris reste fixe
newPos = self.mapToScene(event.pos()) # Position après le zoom
delta = newPos - oldPos
self.translate(delta.x(), delta.y())
except Exception as e:
logger.error(f"Error handling wheel event: {e}", exc_info=True)
def resizeEvent(self, event):
logger.info("Start of resizeEvent in CustomGraphicsView class")
try:
super().resizeEvent(event)
self.resized.emit()
logger.debug("Resize event detected.")
self.adjust_view()
except Exception as e:
logger.error(f"Error in resizeEvent: {e}", exc_info=True)
logger.info("End of resizeEvent in CustomGraphicsView class")
def showEvent(self, event):
logger.info("Start of showEvent in CustomGraphicsView class")
super().showEvent(event)
logger.debug("Show event detected.")
QTimer.singleShot(0, self.adjust_view)
logger.info("End of showEvent in CustomGraphicsView class")
def adjust_view(self):
if self._adjustment_in_progress:
logger.warning("Adjustment already in progress, skipping.")
return
self._adjustment_in_progress = True
try:
if self.scene():
rect_scene = self.scene().itemsBoundingRect()
if not self._initial_view_fitted:
self.fitInView(rect_scene, Qt.KeepAspectRatio)
self._initial_view_fitted = True
logger.debug("Initial view adjusted to show entire scene.")
if not self._signal_connected:
self.scene().changed.connect(self.adjust_view)
self._signal_connected = True
logger.debug("Signal connected to adjust view dynamically.")
except Exception as e:
logger.error(f"Error in adjust_view: {e}", exc_info=True)
finally:
self._adjustment_in_progress = False
self.build_spatial_index() # Construire l'index spatial
def zoom_in(self):
logger.info("Zooming in")
try:
self.scale(self.zoom_in_factor, self.zoom_in_factor)
except Exception as e:
logger.error(f"Error in zoom_in: {e}", exc_info=True)
def zoom_out(self):
logger.info("Zooming out")
try:
self.scale(self.zoom_out_factor, self.zoom_out_factor)
except Exception as e:
logger.error(f"Error in zoom_out: {e}", exc_info=True)
class ControlPoint(QGraphicsEllipseItem):
def __init__(self, position, connected_line=None, parent=None):
self.connected_line = connected_line
super().__init__(-5, -5, 10, 10, parent)
self.setBrush(QBrush(QColor("red")))
self.setFlags(QGraphicsItem.ItemIsMovable | QGraphicsItem.ItemSendsGeometryChanges)
self.setPos(position)
print(f"Création de ControlPoint: {connected_line}") # Vérifier connected_line lors de la création
def itemChange(self, change, value):
try:
connected_line = getattr(self, 'connected_line', None)
logger.debug(f"itemChange - ControlPoint connected_line: {connected_line}")
if change == QGraphicsItem.ItemPositionChange and connected_line:
new_pos = value # 'value' est déjà un QPointF
line = connected_line.line()
if self == connected_line.start_control_point:
connected_line.setLine(QLineF(new_pos, line.p2()))
logger.debug("Updated start_control_point position.")
elif self == connected_line.end_control_point:
connected_line.setLine(QLineF(line.p1(), new_pos))
logger.debug("Updated end_control_point position.")
return new_pos
except Exception as e:
logger.error(f"Error in itemChange: {e}", exc_info=True)
return super().itemChange(change, value)
class MeasuredLine(QGraphicsLineItem):
def __init__(self, line, parent=None):
super().__init__(line, parent)
self.start_control_point = None
self.end_control_point = None
def main():
#profiler = cProfile.Profile()
#profiler.enable()
app = QApplication([])
window = FileViewer()
window.show()
app.exec_()
#profiler.disable()
#s = io.StringIO()
#ps = pstats.Stats(profiler, stream=s).sort_stats('cumulative')
#ps.print_stats()
#logger.debug(f"Profiling results:\n{s.getvalue()}")
if __name__ == "__main__":
main()
# Screen resolution for interface testing: 1920X1080