Source code for umbra.managers.fileSystemEventsManager

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
**fileSystemEventsManager.py**

**Platform:**
	Windows, Linux, Mac Os X.

**Description:**
	Defines the :class:`FileSystemEventsManager` class.

**Others:**

"""

#**********************************************************************************************************************
#***	Future imports.
#**********************************************************************************************************************
from __future__ import unicode_literals

#**********************************************************************************************************************
#***	External imports.
#**********************************************************************************************************************
import os
from PyQt4.QtCore import Qt
from PyQt4.QtCore import QThread
from PyQt4.QtCore import QTimer
from PyQt4.QtCore import pyqtSignal

#**********************************************************************************************************************
#***	Internal imports.
#**********************************************************************************************************************
import foundations.exceptions
import foundations.verbose
import umbra.exceptions
from umbra.globals.constants import Constants

#**********************************************************************************************************************
#***	Module attributes.
#**********************************************************************************************************************
__author__ = "Thomas Mansencal"
__copyright__ = "Copyright (C) 2008 - 2014 - Thomas Mansencal"
__license__ = "GPL V3.0 - http://www.gnu.org/licenses/"
__maintainer__ = "Thomas Mansencal"
__email__ = "[email protected]"
__status__ = "Production"

__all__ = ["LOGGER", "FileSystemEventsManager"]

LOGGER = foundations.verbose.installLogger()

#**********************************************************************************************************************
#***	Module classes and definitions.
#**********************************************************************************************************************
[docs]class FileSystemEventsManager(QThread):
	"""
	Defines the file system events manager.
	"""

	# Custom signals definitions.
	fileChanged = pyqtSignal(unicode)
	"""
	This signal is emited by the :class:`FileSystemEventsManager` class when a file is changed. ( pyqtSignal )

	:return: Current changed file.
	:rtype: unicode
	"""

	fileInvalidated = pyqtSignal(unicode)
	"""
	This signal is emited by the :class:`FileSystemEventsManager` class when a file is invalidated. ( pyqtSignal )

	:return: Current invalidated file.
	:rtype: unicode
	"""

	directoryChanged = pyqtSignal(unicode)
	"""
	This signal is emited by the :class:`FileSystemEventsManager` class when a directory is changed. ( pyqtSignal )

	:return: Current changed directory.
	:rtype: unicode
	"""

	directoryInvalidated = pyqtSignal(unicode)
	"""
	This signal is emited by the :class:`FileSystemEventsManager` class when a directory is invalidated. ( pyqtSignal )

	:return: Current invalidated directory.
	:rtype: unicode
	"""

	def __init__(self, parent=None):
		"""
		Initializes the class.
		"""

		LOGGER.debug("> Initializing '{0}()' class.".format(self.__class__.__name__))

		QThread.__init__(self, parent)

		# --- Setting class attributes. ---
		self.__container = parent

		self.__paths = {}

		self.__timer = None
		self.__timerCycleMultiplier = 5

		self.__toRegisterPaths = {}
		self.__toUnregisterPaths = []

	#******************************************************************************************************************
	#***	Attributes properties.
	#******************************************************************************************************************
	@property
	def container(self):
		"""
		Property for **self.__container** attribute.

		:return: self.__container.
		:rtype: QObject
		"""

		return self.__container

	@container.setter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
	def container(self, value):
		"""
		Setter for **self.__container** attribute.

		:param value: Attribute value.
		:type value: QObject
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is read only!".format(self.__class__.__name__, "container"))

	@container.deleter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
[docs]	def container(self):
		"""
		Deleter for **self.__container** attribute.
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is not deletable!".format(self.__class__.__name__, "container"))

	@property
	def paths(self):
		"""
		Property for **self.__paths** attribute.

		:return: self.__paths.
		:rtype: dict
		"""

		return self.__paths

	@paths.setter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
	def paths(self, value):
		"""
		Setter for **self.__paths** attribute.

		:param value: Attribute value.
		:type value: dict
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is read only!".format(self.__class__.__name__, "paths"))

	@paths.deleter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
[docs]	def paths(self):
		"""
		Deleter for **self.__paths** attribute.
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is not deletable!".format(self.__class__.__name__, "paths"))

	@property
	def timer(self):
		"""
		Property for **self.__timer** attribute.

		:return: self.__timer.
		:rtype: QTimer
		"""

		return self.__timer

	@timer.setter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
	def timer(self, value):
		"""
		Setter for **self.__timer** attribute.

		:param value: Attribute value.
		:type value: QTimer
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is read only!".format(self.__class__.__name__, "timer"))

	@timer.deleter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
[docs]	def timer(self):
		"""
		Deleter for **self.__timer** attribute.
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is not deletable!".format(self.__class__.__name__, "timer"))

	@property
	def timerCycleMultiplier(self):
		"""
		Property for **self.__timerCycleMultiplier** attribute.

		:return: self.__timerCycleMultiplier.
		:rtype: float
		"""

		return self.__timerCycleMultiplier

	@timerCycleMultiplier.setter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
	def timerCycleMultiplier(self, value):
		"""
		Setter for **self.__timerCycleMultiplier** attribute.

		:param value: Attribute value.
		:type value: float
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is read only!".format(self.__class__.__name__, "timerCycleMultiplier"))

	@timerCycleMultiplier.deleter
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.ProgrammingError)
[docs]	def timerCycleMultiplier(self):
		"""
		Deleter for **self.__timerCycleMultiplier** attribute.
		"""

		raise foundations.exceptions.ProgrammingError(
		"{0} | '{1}' attribute is not deletable!".format(self.__class__.__name__, "timerCycleMultiplier"))

	#******************************************************************************************************************
	#***	Class methods.
	#******************************************************************************************************************
	def __getitem__(self, path):
		"""
		Reimplements the :meth:`object.__getitem__` method.

		:param path: Path name.
		:type path: unicode
		:return: Path.
		:rtype: Path
		"""

		return self.__paths.__getitem__(path)

	def __setitem__(self, path, modifiedTime):
		"""
		Reimplements the :meth:`object.__setitem__` method.

		:param path: Path.
		:type path: unicode
		:param modifiedTime: Modified time.
		:type modifiedTime: int or float
		"""

		self.registerPath(path, modifiedTime)

	def __iter__(self):
		"""
		Reimplements the :meth:`object.__iter__` method.

		:return: Paths iterator.
		:rtype: object
		"""

		return self.__paths.iteritems()

	def __contains__(self, path):
		"""
		Reimplements the :meth:`object.__contains__` method.

		:param path: Path name.
		:type path: unicode
		:return: Path existence.
		:rtype: bool
		"""

		return path in self.__paths

	def __len__(self):
		"""
		Reimplements the :meth:`object.__len__` method.

		:return: Paths count.
		:rtype: int
		"""

		return len(self.__paths)

[docs]	def get(self, path, default=None):
		"""
		Returns given path value.

		:param path: Path name.
		:type path: unicode
		:param default: Default value if path is not found.
		:type default: object
		:return: Action.
		:rtype: QAction
		"""

		try:
			return self.__getitem__(path)
		except KeyError as error:
			return default

[docs]	def run(self):
		"""
		Reimplements the :meth:`QThread.run` method.
		"""

		self.__timer = QTimer()
		self.__timer.moveToThread(self)
		self.__timer.start(Constants.defaultTimerCycle * self.__timerCycleMultiplier)

		self.__timer.timeout.connect(self.__watchFileSystem, Qt.DirectConnection)

		self.exec_()

	def __watchFileSystem(self):
		"""
		Watches the file system for paths that have been changed or invalidated on disk.
		"""

		for path, data in self.__paths.items():
			storedModifiedTime, isFile = data
			try:
				if not foundations.common.pathExists(path):
					LOGGER.warning(
					"!> {0} | '{1}' path has been invalidated and will be unregistered!".format(self.__class__.__name__, path))
					del(self.__paths[path])
					if isFile:
						self.fileInvalidated.emit(path)
					else:
						self.directoryInvalidated.emit(path)
					continue
			except KeyError:
				LOGGER.debug("> {0} | '{1}' path has been unregistered while iterating!".format(
				self.__class__.__name__, path))
				continue

			try:
				modifiedTime = self.getPathModifiedTime(path)
			except OSError:
				LOGGER.debug("> {0} | '{1}' path has been invalidated while iterating!".format(
				self.__class__.__name__, path))
				continue

			if storedModifiedTime != modifiedTime:
				self.__paths[path] = (modifiedTime, os.path.isfile(path))
				LOGGER.debug("> {0} | '{1}' path has been changed!".format(self.__class__.__name__, path))
				if isFile:
					self.fileChanged.emit(path)
				else:
					self.directoryChanged.emit(path)

[docs]	def listPaths(self):
		"""
		Returns the registered paths.

		:return: Registered paths.
		:rtype: list
		"""

		return sorted(self.__paths.keys())

[docs]	def isPathRegistered(self, path):
		"""
		Returns if the given path is registered.

		:param path: Path name.
		:type path: unicode
		:return: Is path registered.
		:rtype: bool
		"""

		return path in self
# Oncilla: Statement commented by auto-documentation process: 
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(foundations.exceptions.PathExistsError,
# Oncilla: Statement commented by auto-documentation process: 											umbra.exceptions.PathRegistrationError)
[docs]	def registerPath(self, path, modifiedTime=None):
		"""
		Registers given path.

		:param path: Path name.
		:type path: unicode
		:param modifiedTime: Custom modified time.
		:type modifiedTime: int or float
		:return: Method success.
		:rtype: bool
		"""

		if not foundations.common.pathExists(path):
			raise foundations.exceptions.PathExistsError("{0} | '{1}' path doesn't exists!".format(
			self.__class__.__name__, path))

		if path in self:
			raise umbra.exceptions.PathRegistrationError("{0} | '{1}' path is already registered!".format(
			self.__class__.__name__, path))

		self.__paths[path] = (self.getPathModifiedTime(path) if modifiedTime is None else modifiedTime, os.path.isfile(path))
		return True
# Oncilla: Statement commented by auto-documentation process: 
# Oncilla: Statement commented by auto-documentation process: 	@foundations.exceptions.handleExceptions(umbra.exceptions.PathExistsError)
[docs]	def unregisterPath(self, path):
		"""
		Unregisters given path.

		:param path: Path name.
		:type path: unicode
		:return: Method success.
		:rtype: bool
		"""

		if not path in self:
			raise umbra.exceptions.PathExistsError("{0} | '{1}' path isn't registered!".format(
			self.__class__.__name__, path))

		del(self.__paths[path])
		return True
# Oncilla: Statement commented by auto-documentation process: 
# Oncilla: Statement commented by auto-documentation process: 	@staticmethod
[docs]	def getPathModifiedTime(path):
		"""
		Returns given path modification time.

		:param path: Path.
		:type path: unicode
		:return: Modification time.
		:rtype: int
		"""

		return float(foundations.common.getFirstItem(str(os.path.getmtime(path)).split(".")))