Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# to the system
if pygtftk.__NON_INTERACTIVE__:
message("Error encountered. System will exit after deleting temporary files.", type="DEBUG")
if not pygtftk.__ARGS__['keep_all']:
for i in flatten_list(TMP_FILE_LIST, outlist=[]):
message("ERROR encountered, deleting temporary file: " + i, type="DEBUG")
silentremove(i)
else:
message("Deletion of temporary files canceled by user.", type="DEBUG")
sys.exit(1)
else:
raise GTFtkInteractiveError(msg)
if force:
pygtftk.utils.VERBOSITY = VERBOSITY_BACK
def message(msg, nl=True, type="INFO", force=False):
"""Send a formated message on STDERR.
:param msg: the message to be send.
:param nl: if True, add a newline.
:param type: Could be INFO, ERROR, WARNING.
:param force: Force message, whatever the verbosity level.
>>> from pygtftk.utils import message
"""
now = datetime.datetime.now()
if pygtftk.utils.VERBOSITY > 2:
ho_min = str(now.hour) + ":" + str(now.minute).zfill(2) + ":" + str(now.second).zfill(2)
else:
ho_min = str(now.hour) + ":" + str(now.minute).zfill(2)
do_it = False
if type not in ["INFO", "ERROR", "WARNING", "DEBUG", "DEBUG_MEM"]:
raise GTFtkError(
"Type should be one of INFO, ERROR, WARNING, DEBUG, DEBUG_MEM.")
if force:
VERBOSITY_BACK = pygtftk.utils.VERBOSITY
pygtftk.utils.VERBOSITY = 2
if pygtftk.utils.VERBOSITY >= 2:
do_it = True
'txt': txt_regexp,
'txt.gz': txt_regexp_gz,
'bigwig': bigwig_regexp,
'zip': zip_regexp,
'pdf': pdf_regexp}
# Set verbosity system wide as depending on
# command line argument order, VERBOSITY (-V) can
# be evaluated later...
if '-V' in sys.argv:
sys_args = ' '.join(sys.argv)
verbosity_val = re.search('-V ?([01234])?', sys_args)
if verbosity_val:
pygtftk.utils.VERBOSITY = int(verbosity_val.group(1))
else:
pygtftk.utils.VERBOSITY = 0
match = False
if isinstance(self.file_ext, str):
extension_list = [self.file_ext]
else:
extension_list = list(self.file_ext)
for this_ext in extension_list:
if re.search(ext2regexp[this_ext], string):
match = True
break
if not match:
message('Not a valid filename extension :' + string, type="WARNING")
message('Extension expected: ' + ext2regexp[this_ext], type="ERROR")
now = datetime.datetime.now()
if pygtftk.utils.VERBOSITY > 2:
ho_min = str(now.hour) + ":" + str(now.minute).zfill(2) + ":" + str(now.second).zfill(2)
else:
ho_min = str(now.hour) + ":" + str(now.minute).zfill(2)
do_it = False
if type not in ["INFO", "ERROR", "WARNING", "DEBUG", "DEBUG_MEM"]:
raise GTFtkError(
"Type should be one of INFO, ERROR, WARNING, DEBUG, DEBUG_MEM.")
if force:
VERBOSITY_BACK = pygtftk.utils.VERBOSITY
pygtftk.utils.VERBOSITY = 2
if pygtftk.utils.VERBOSITY >= 2:
do_it = True
elif pygtftk.utils.VERBOSITY == 1:
if type == "ERROR":
do_it = True
elif type == "WARNING":
do_it = True
elif type == "INFO":
do_it = True
elif type == "DEBUG_MEM":
do_it = False
elif type == "DEBUG":
do_it = False
# -------------------------------------------------------------------------
if ensembl_collection == 'vertebrate':
host = "ftp.ensembl.org"
user = "anonymous" # votre identifiant
password = "anonymous@gtftk.fr"
elif ensembl_collection in ['protists', 'fungi', 'plants', 'metazoa']:
host = "ftp.ensemblgenomes.org"
user = "anonymous"
password = "anonymous@gtftk.fr"
try:
ftp = ftputil.FTPHost(host, user, password)
if pygtftk.utils.VERBOSITY:
message("Connected to ensembl FTP website.")
except FTPOSError as err:
message(str(err))
message("Unable to connect (FTPOSError).", type="ERROR")
try:
ftp.chdir('/pub')
message("Successfully change directory to pub")
except:
message("Unable to change directory to 'pub'.",
type="ERROR")
if ensembl_collection in ['protists', 'fungi', 'plants', 'metazoa']:
try:
ftp.chdir(ensembl_collection)
message("Successfully change directory to " + ensembl_collection)
# -------------------------------------------------------------------------
# Compute coverage
#
# -------------------------------------------------------------------------
result_bed = bw_cov_mp(bw_list=bw_list,
region_file=open(region_bed.name),
labels=labels,
bin_nb=nb_window,
pseudo_count=pseudo_count,
zero_to_na=zero_to_na,
nb_proc=nb_proc,
n_highest=n_highest,
stat=stat,
verbose=pygtftk.utils.VERBOSITY)
if matrix_out:
result_bed.close()
df_first = pd.read_csv(result_bed.name, sep="\t", header=None)
if int(re.sub("\..*", "", pd.__version__)) > 0:
df_first = df_first.iloc[:, [0, 1, 2, 3, 5, 4]]
else:
df_first = df_first.ix[:, [0, 1, 2, 3, 5, 4]]
df_list = []
for i in range(len(labels)):
# create a sub data frame containing the coverage values of the
# current bwig
for key, value in list(args.items()):
if isinstance(value, io.IOBase):
value = value.name
else:
value = str(value)
cmd_ob.logger.info("Argument: " + key + "=" + value)
# Set the level of verbosity
# Can be None if -V is used without value
# (nargs)
if args['verbosity'] is None:
pygtftk.utils.VERBOSITY = 1
else:
pygtftk.utils.VERBOSITY = int(args['verbosity'])
# Set whether date should be added to
# output file
if args['no_date']:
pygtftk.utils.ADD_DATE = False
arg_list_to_del = ['bash_comp', 'add_chr', 'version', 'help',
'plugin_tests', 'list_plugins', 'plugin_tests_no_conn',
'r_libs', 'add_plugin', 'update_plugins', 'system_info',
'plugin_path', 'no_date', 'keep_all', 'logger_file',
'tmp_dir', 'verbosity', 'command', 'write_message_to_file']
for cur_arg in arg_list_to_del:
try:
del args[cur_arg]
except KeyError:
pass
now = datetime.datetime.now()
if pygtftk.utils.VERBOSITY > 2:
ho_min = str(now.hour) + ":" + str(now.minute).zfill(2) + ":" + str(now.second).zfill(2)
else:
ho_min = str(now.hour) + ":" + str(now.minute).zfill(2)
do_it = False
if type not in ["INFO", "ERROR", "WARNING", "DEBUG", "DEBUG_MEM"]:
raise GTFtkError(
"Type should be one of INFO, ERROR, WARNING, DEBUG, DEBUG_MEM.")
if force:
VERBOSITY_BACK = pygtftk.utils.VERBOSITY
pygtftk.utils.VERBOSITY = 2
if pygtftk.utils.VERBOSITY >= 2:
do_it = True
elif pygtftk.utils.VERBOSITY == 1:
if type == "ERROR":
do_it = True
elif type == "WARNING":
do_it = True
elif type == "INFO":
do_it = True
elif type == "DEBUG_MEM":
do_it = False
elif type == "DEBUG":
do_it = False
suffix=".bed")
dws_region_bo.saveas(dws_bed_file.name)
tmp_file = bw_profile_mp(in_bed_file=dws_bed_file.name,
nb_proc=nb_proc,
big_wig=[
x.name for x in bigwiglist],
bin_nb=around_bin_nb,
pseudo_count=pseudo_count,
stranded=not no_stranded,
type="downstream",
labels=labels,
outputfile=outputfile.name,
zero_to_na=zero_to_na,
verbose=pygtftk.utils.VERBOSITY)
outputfile_list["downstream"] = tmp_file
# -------------------------------------------------------------------------
#
# Merge file using pandas
#
# -------------------------------------------------------------------------
message("Reading (pandas): " + outputfile_list["main"].name, type="DEBUG")
df_main = pd.read_csv(outputfile_list["main"].name, sep="\t")
# save strand and end
# They will re-joined added later
df_copy = df_main[['bwig', 'chrom', 'gene',
'strand', 'start', 'end']]