Source code for ecpi.common

"""generic classes, functions. Could be used by other projects.
"""

import os.path as osp
import os
import shutil
from glob import glob
# WorkSpace directories
workspace_dirs = {
    'tests': 'DATA_TESTS',
    'process': 'DATA_PROCESS',
    'output': 'OUTPUT'
}


[docs]class DirForTest: """create output directory with same path than test module """ def __init__(self, p_module_path): """ :param p_module_path: """ self.sep = osp.sep base = 'ecpi_garage' if p_module_path.find(base) < 0: base = 'ecpi' # *basedir* must contain the last separator self.sep # resulting *p_module_path* must start without path separator basedir = osp.join(get_root_eclairs(), base) + self.sep p_tests = self.sep + 'tests' p_module_path = p_module_path.replace(basedir, '').replace(p_tests, '').replace('.py', '') print(p_module_path) self.path_asso = add_path_tests_eclairs(p_module_path) if not osp.isdir(self.path_asso): os.makedirs(self.path_asso) def __call__(self, p_file): """ function to define a path :param p_file: path of file :return: a path at the same level that p_module :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == self.sep: p_file = p_file[1:] return osp.join(self.path_asso, p_file)
[docs] def is_file(self, p_file): """Returns if p_file file exists in ref path. :param p_module: path of module :param p_file: path of file """ return osp.exists(self.__call__(p_file))
[docs] def rm_all(self): """remove all files """ if osp.isdir(self.path_asso): shutil.rmtree(self.path_asso) os.makedirs(self.path_asso)
[docs] def rm_all_fits(self): """remove all FITS files """ for file in glob(osp.join(self.path_asso, "*.fits")): os.remove(file)
[docs]def add_path_current_module(p_module, p_file): """ function to define a path :param p_module: path of module, ie __file__ :param p_file: path of file :return: a path at the same level that p_module :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(osp.dirname(p_module), p_file)
[docs]def get_root_eclairs(): """get the path of the ECLAIRs directory. :return: path of the root of the ECLAIRs directory :rtype: string """ # Checking if we are in a git cloned folder or in a local installation root = os.getenv('ECPI_ROOT') if not root: l_sep = osp.sep full = __file__.split(l_sep) root = l_sep.join(full[:-3]) return root
[docs]def get_home_eclairs(): """get the path of the ECPI_HOMEs directory. :return: path of the home of the ECLAIRs directory :rtype: string """ # Checking if we are in a git cloned folder or in a local installation home = os.getenv('ECPI_HOME') if not home and __file__.find('ECLAIRs') >= 0: l_sep = osp.sep full = __file__.split(l_sep) home = l_sep.join(full[:-4]) return home
[docs]def get_caldb_dir(): """get the path of the ECLAIRs directory. :return: path of the root of the ECLAIRs directory :rtype: string """ # Checking if we are in a git cloned folder or in a local installation return osp.join(get_root_eclairs(), "ecpi", "extras", "caldb")
[docs]def get_path_pipeline(): """gets the path of the pipeline's home directory. :return: absolute path to pipeline's folder :rtype: string """ return osp.join(get_root_eclairs(), 'ecpi', 'pipeline')
[docs]def add_path_home_eclairs(p_file): """add to p_file the path of the ECLAIRs directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_home_eclairs(), p_file)
[docs]def add_path_root_eclairs(p_file): """add to p_file the path of the ECLAIRs directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_root_eclairs(), p_file)
[docs]def add_path_caldb_eclairs(p_file): """add to p_file the path of the CalDB directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_caldb_dir(), p_file)
[docs]def add_path_tests_eclairs(p_file): """add to p_file the path of the ECLAIRs test directory. The ECLAIRs test directory is ECLAIRs/DATA_TEST. It is a non committed directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_home_eclairs(), workspace_dirs['tests'], p_file)
[docs]def add_path_process_eclairs(p_file): """add to p_file the path of the ECLAIRs processing directory. The ECLAIRs test directory is ECLAIRs/DATA_PROCESS. It is a non committed directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_home_eclairs(), workspace_dirs['process'], p_file)
[docs]def add_path_output_eclairs(p_file): """add to p_file the path of the ECLAIRs output directory. The ECLAIRs output directory is ECLAIRs/OUTPUT. It is a non commited directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_home_eclairs(), workspace_dirs['output'], p_file)
[docs]def add_path_data_ref_eclairs(p_file): """add to p_file the path of the ECLAIRs data simulation or analysis directory. :return: absolute path of p_file :rtype: string """ if p_file.find(osp.sep) > 0: if p_file[0] == osp.sep: p_file = p_file[1:] return osp.join(get_root_eclairs(), 'ecpi', 'common', 'data', p_file)
[docs]def get_path_associated_test(): """add to p_file the path of the ECLAIRs output directory. The ECLAIRs output directory is ECLAIRs/OUTPUT. It is a non committed directory. :return: absolute path of p_file :rtype: string """ pass