Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self._snps.empty:
self._source.extend(source)
self._snps = snps
else:
common_snps = self._snps.join(snps, how="inner", rsuffix="_added")
discrepant_positions = common_snps.loc[
(common_snps["chrom"] != common_snps["chrom_added"])
| (common_snps["pos"] != common_snps["pos_added"])
]
if not self._name:
prefix = ""
else:
prefix = "{}_".format(clean_str(self._name))
if 0 < len(discrepant_positions) < discrepant_snp_positions_threshold:
logger.warning(
"{} SNP positions were discrepant; keeping original positions".format(
str(len(discrepant_positions))
)
)
if save_output:
self._discrepant_positions_file_count += 1
save_df_as_csv(
discrepant_positions,
self._output_dir,
"{}discrepant_positions_{}{}".format(
prefix, str(self._discrepant_positions_file_count), ".csv"
),
def _save_discrepant_snps_file(self, df, discrepant_snps_type, filename):
if not filename:
if not self._name:
filename = "{}.csv".format(discrepant_snps_type)
else:
filename = "{}_{}.csv".format(
clean_str(self._name), discrepant_snps_type
)
return save_df_as_csv(
df,
self._output_dir,
filename,
comment="# Source(s): {}\n".format(self.source),
)
if self._snps.empty:
self._source.extend(source)
self._snps = snps
else:
common_snps = self._snps.join(snps, how="inner", rsuffix="_added")
discrepant_positions = common_snps.loc[
(common_snps["chrom"] != common_snps["chrom_added"])
| (common_snps["pos"] != common_snps["pos_added"])
]
if not self._name:
prefix = ""
else:
prefix = "{}_".format(clean_str(self._name))
if 0 < len(discrepant_positions) < discrepant_snp_positions_threshold:
print(
"{} SNP positions were discrepant; keeping original positions".format(
str(len(discrepant_positions))
)
)
if save_output:
self._discrepant_positions_file_count += 1
save_df_as_csv(
discrepant_positions,
self._output_dir,
"{}discrepant_positions_{}{}".format(
prefix, str(self._discrepant_positions_file_count), ".csv"
),
if self._snps.empty:
self._source.extend(source)
self._snps = snps
else:
common_snps = self._snps.join(snps, how="inner", rsuffix="_added")
discrepant_positions = common_snps.loc[
(common_snps["chrom"] != common_snps["chrom_added"])
| (common_snps["pos"] != common_snps["pos_added"])
]
if not self._name:
prefix = ""
else:
prefix = "{}_".format(clean_str(self._name))
if 0 < len(discrepant_positions) < discrepant_snp_positions_threshold:
logger.warning(
"{} SNP positions were discrepant; keeping original positions".format(
str(len(discrepant_positions))
)
)
if save_output:
self._discrepant_positions_file_count += 1
save_df_as_csv(
discrepant_positions,
self._output_dir,
"{}discrepant_positions_{}{}".format(
prefix, str(self._discrepant_positions_file_count), ".csv"
),
def _save_discrepant_snps_file(self, df, discrepant_snps_type, filename):
if not filename:
if not self._name:
filename = "{}.csv".format(discrepant_snps_type)
else:
filename = "{}_{}.csv".format(
clean_str(self._name), discrepant_snps_type
)
return save_df_as_csv(
df,
self._output_dir,
filename,
comment="# Source(s): {}\n".format(self.source),
)
def _save_discrepant_snps_file(self, df, discrepant_snps_type, filename):
if not filename:
if not self._name:
filename = "{}.csv".format(discrepant_snps_type)
else:
filename = "{}_{}.csv".format(
clean_str(self._name), discrepant_snps_type
)
return save_df_as_csv(
df,
self._output_dir,
filename,
comment="# Source(s): {}\n".format(self.source),
)
vcf : bool
flag to save file as VCF
atomic : bool
atomically write output to a file on local filesystem
**kwargs
additional parameters to `pandas.DataFrame.to_csv`
Returns
-------
str
path to file in output directory if SNPs were saved, else empty str
"""
if not self._name:
prefix = ""
else:
prefix = "{}_".format(clean_str(self._name))
if not filename:
if vcf:
ext = ".vcf"
else:
ext = ".csv"
filename = "{}{}{}".format(prefix, self.assembly, ext)
return super().save_snps(filename=filename, vcf=vcf, atomic=atomic, **kwargs)