Commit ebd65a03 authored by eric pellegrini's avatar eric pellegrini
Browse files

started to repopulate MDANSE

parent c774502c
import abc
class _Meta(type):
'''
Metaclass that allows to use the __getitem__ method at a class level for the class that has been built.
The class that uses this metaclass must define a class attribute named _registry that will be used
by the __getitem__ method.
'''
def __getitem__(self, item):
"""
Returns a given item stored in the class registry
"""
return self._registry[item]
class ClassRegistry(abc.ABCMeta):
'''
Metaclass that registers the subclasses of bases classes.
The internal registry is defined as a nested dictionary whose keys
are the |type| class attribute of the base classes and values another dictionary
whose keys are the |type| class attribute of the subclasses and values are the corresponding
class instances.
Hence any base or child class that does not define |type| class attribute will not be registered.
'''
__metaclass__ = _Meta
__interfaces = []
_registry = {}
def __init__(self, name, bases, namespace):
'''
Constructor of a class metaclassed by ClassFactory
:param name: the name of the class to be built by this metaclass
:param bases: the base classes of the class to be built by this metaclass
:param namespace: the attributes and methods of the class to be built by this metaclass
'''
super(ClassRegistry, self).__init__(name, bases, namespace)
# Get the typ of the class
typ = getattr(self, 'type', None)
if typ is None:
return
metaClass = namespace.get("__metaclass__", None)
if metaClass is ClassRegistry:
ClassRegistry.__interfaces.append(self)
ClassRegistry._registry[typ] = {}
else:
for interface in ClassRegistry.__interfaces:
if issubclass(self, interface):
ClassRegistry._registry[interface.type][typ] = self
break
@classmethod
def info(cls, interface):
'''
Returns informations about the subclasses of a given base class stored in the registry.
:param cls: the ClassRegsitry instance
:param interface: the name of base class of whom information about its subclasses is requested
'''
if not cls._registry.has_key(interface):
return "The interface " + interface + " is not registered"
import inspect
import os
# Dictionnay whose keys are the package names and values and list of (job name, job path) stored in the corresponding package.
packages = {}
# Loop over the registry items.
for k, v in cls._registry[interface].items():
# Get the module corresponding to the job class.
mod = inspect.getmodule(v)
# The package hosting the module.
modPackage = mod.__package__
# The module file.
modFilename = mod.__file__
# If no package could be found, guess a name using the directory name of the module file.
if modPackage is None:
modPackage = os.path.split(os.path.dirname(modFilename))[1]
# Update the packages dictionary.
if packages.has_key(modPackage):
packages[modPackage].append([k, v.__name__])
else:
packages[modPackage] = [[k, v.__name__]]
contents = []
# Print the contents of the packages dictionary.
contents.append("="*130)
contents.append("%-50s %-40s %-s" % ("Package", "Name", "Class"))
contents.append("="*130)
for k, v in sorted(packages.items()):
for vv in sorted(v):
contents.append("%-50s %-40s %-s" % (k, vv[0], vv[1]))
contents.append('-' * 130)
contents = "\n".join(contents)
return contents
"""
This class implements the MDANSE error handler.
It inherits from the Exception class. It can be used to filter the exceptions
raised within the MDANSE framework
"""
class Error(Exception):
def __init__(self, msg=None):
self._msg = msg
def __str__(self):
return repr(self._msg)
import logging
import sys
class Logger(object):
levels = {"debug" : logging.DEBUG,
"info" : logging.INFO,
"warning" : logging.WARNING,
"error" : logging.ERROR,
"fatal" : logging.CRITICAL,
"critical" : logging.FATAL
}
def __call__(self, message, level="info", loggers=None):
lvl = Logger.levels.get(level, None)
# If the logging level is unkwnown, skip that log
if lvl is None:
return
if loggers is None:
loggers = logging.Logger.manager.loggerDict.keys()
else:
loggers = [n for n in loggers if logging.Logger.manager.loggerDict.has_key(n)]
for n in loggers:
logging.getLogger(n).log(lvl, message)
def start(self, logger=None):
from nMOLDYN import _nmoldyn_excepthook
sys.excepthook = _nmoldyn_excepthook
if loggers is None:
loggers = logging.Logger.manager.loggerDict.keys()
else:
loggers = [n for n in loggers if logging.Logger.manager.loggerDict.has_key(n)]
for n in loggers:
logging.getLogger(n).disabled = False
def stop(self, loggers=None):
sys.excepthook = _sys_excepthook
if loggers is None:
loggers = logging.Logger.manager.loggerDict.keys()
else:
loggers = [n for n in loggers if logging.Logger.manager.loggerDict.has_key(n)]
for n in loggers:
logging.getLogger(n).disabled = True
def set_level(self, level, loggers=None):
lvl = Logger.levels.get(level, None)
if lvl is None:
return
if loggers is None:
loggers = logging.Logger.manager.loggerDict.keys()
else:
loggers = [n for n in loggers if logging.Logger.manager.loggerDict.has_key(n)]
for loggerName in loggers:
logging.getLogger(loggerName).setLevel(lvl)
def add_logger(self, name, handler, level="error"):
if logging.Logger.manager.loggerDict.has_key(name):
return
logging.getLogger(name).addHandler(handler)
self.set_level(level, loggers=[name])
"""
This modules implements actions or informations that are platform-specific.
"""
# Standards imports.
import abc
import ctypes
import datetime
import getpass
import os
import re
import subprocess
from MDANSE.Core.Error import Error
class PlatformError(Error):
'''
Handles error occuring in the module.
'''
pass
class Platform(object):
"""Virtual base class for platform-specific methods.
@note: this class is designed according to the singleton pattern.
"""
__metaclass__ = abc.ABCMeta
__instance = None
def __new__(cls, *args, **kwargs):
'''
Create a new instance of Platform class.
@param cls: the class to instantiate.
@type cls: class
@note: designed using the Singleton pattern.
'''
# Case of the first instantiation.
if cls.__instance is None:
cls.__instance = super(Platform, cls).__new__(cls, *args, **kwargs)
# The selected instance is returned.
return cls.__instance
@abc.abstractmethod
def application_directory(self):
"""Returns the path for MDANSE application directory.
@return: the path for MDANSE application directory.
@rtype: string
@attention: this is an abstract method.
"""
pass
def documentation_path(self):
return os.path.join(self.package_directory(), 'GUI', 'Help')
def local_mmtk_database_directory(self):
path = os.path.join(self.application_directory(), 'mmtk_database')
if not os.path.exists(path):
for mmtkDatabaseType in ['Atoms', 'Complexes', 'Crystals', 'Groups', 'Molecules', 'Proteins']:
os.makedirs(os.path.join(path, mmtkDatabaseType))
return path
def change_directory(self, directory):
os.chdir(directory)
def is_directory_writable(self, path, testFile="junk.test.xxx"):
path = self.get_path(path)
if not os.path.exists(path):
# Try to make the directory.
try:
os.makedirs(path)
# An error occured.
except OSError:
return False
testFile = os.path.join(path, testFile)
try:
f = open(testFile, "w")
except IOError:
return False
else:
f.close()
os.unlink(testFile)
return True
def is_file_writable(self, path, delete=True):
path = self.get_path(path)
if not self.is_directory_writable(os.path.dirname(path)):
return False
try:
f = open(path, "w")
except IOError:
return False
else:
f.close()
if delete:
os.unlink(path)
return True
def create_directory(self, path):
path = self.get_path(path)
if os.path.exists(path):
return
# Try to make the directory.
try:
os.makedirs(path)
# An error occured.
except OSError as e:
raise PlatformError(e)
def get_path(self, path):
path = str(path).encode('string-escape')
path = os.path.abspath(os.path.expanduser(path))
return path
def database_default_path(self):
'''Returns the path for default mdanse database.
@return: the mdanse database path.
@rtype: string
'''
return os.path.join(self.package_directory(), 'Data', 'elements_database.csv')
def database_user_path(self):
'''Returns the path for user mdanse database.
@return: the mdanse database path.
@rtype: string
'''
return os.path.join(self.application_directory(), 'elements_database.csv')
def database_path(self):
'''Returns the path for mdanse database.
@return: the mdanse database path.
@rtype: string
'''
path = self.database_user_path()
if os.path.exists(path):
return path
else:
return self.database_default_path()
@abc.abstractmethod
def get_processes_info(self):
'''Returns the active processes.
@return: a mapping between active processes pid and their corresponding process name.
@rtype: dict
@attention: this is an abstract method.
'''
pass
@abc.abstractmethod
def kill_process(self, pid):
"""Kill the specified process.
@param process: the pid of the process to be killed.
@type process: integer
@attention: this is an abstract method.
"""
pass
def pid(self):
return os.getpid()
def package_directory(self):
"""Returns the path for mdanse package.
@return: the path for mdanse package.
@rtype: string
"""
return os.path.dirname(os.path.dirname(__file__))
@abc.abstractmethod
def preferences_file(self):
"""Filename of our preferences file.
@return: the filename of the mdanse preference file.
@rtype: string
@attention: this is an abstract method.
"""
pass
def standard_jobs_directory(self):
'''Returns the mdanse jobs directory.
@return: the mdanse jobs directory.
@rtype: string
'''
basedir = self.package_directory()
return os.path.join(basedir, 'Framework', 'Jobs')
def temporary_files_directory(self):
'''Returns the mdanse temporary files directory.
It will contains the files containing the information about a running job.
@return: the mdanse temporary files directory.
@rtype: string
'''
path = os.path.join(self.application_directory(), 'temporary_files')
self.create_directory(path)
return path
def username(self):
'''Returns the username for the running mdanse session..
@return: the username.
@rtype: string
'''
return getpass.getuser().lower()
class PlatformPosix(Platform):
"""Common Platform base class for Linux and Mac."""
__metaclass__ = abc.ABCMeta
def home_directory(self):
"""Returns the home directory.
@return: the home directory
@rtype: string
"""
return os.environ['HOME']
def kill_process(self, pid):
"""Kill the specified process.
@param process: the pid of the process to be killed.
@type process: integer
"""
import signal
os.kill(pid, signal.SIGTERM)
def application_directory(self):
"""Returns the path for mdanse application directory.
@return: the path for mdanse application directory.
@rtype: string
"""
basedir = os.path.join(os.environ['HOME'], '.mdanse')
# If the application directory does not exist, create it.
if not os.path.exists(basedir):
os.makedirs(basedir)
return basedir
def rename(self, src, dst):
os.rename(src, dst)
def preferences_file(self):
"""Filename of our preferences file.
@return: the filename of the mdanse preference file.
@rtype: string
"""
# The preferences files will be located in the application directory.
appdir = self.application_directory()
return os.path.join(appdir, 'mdanse_preferences')
def etime_to_ctime(self, etime):
etime = [0, 0, 0] + [int(v) for v in re.split("-|:", etime)]
days, hours, minutes, seconds = etime[-4:]
etime = datetime.timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
return (datetime.datetime.today() - etime).strftime("%d-%m-%Y %H:%M:%S")
def get_processes_info(self):
'''Returns the active processes.
@return: a mapping between active processes pid and their corresponding process name.
@rtype: dict
'''
# Get all the active processes using the Unix ps command
procs = subprocess.Popen(['ps', '-o', 'pid,etime'], stdout=subprocess.PIPE)
# The output of the ps command is splitted according to line feeds.
procs = procs.communicate()[0].split('\n')[1:]
# The list of (pid,executable).
procs = [p.split() for p in procs if p]
# A mapping between the active processes pid and their corresponding exectuable.
procs = dict([(int(p[0].strip()), self.etime_to_ctime(p[1].strip())) for p in procs])
return procs
class PlatformMac(PlatformPosix):
"""
Mac-specific platform object.
"""
name = "macos"
class PlatformLinux(PlatformPosix):
"""
Linux-specific platform object.
"""
name = "linux"
class PlatformWin(Platform):
"""
Win-specific platform object.
"""
name = "windows"
def application_directory(self):
"""Returns the path for mdanse application directory.
@return: the path for mdanse application directory.
@rtype: string
"""
basedir = os.path.join(os.environ['APPDATA'], 'mdanse')
# If the application directory does not exist, create it.
if not os.path.exists(basedir):
os.makedirs(basedir)
return basedir
def get_process_creation_time(self, process):
creationtime = ctypes.c_ulonglong()