本文整理汇总了Python中qiita_db.metadata_template.PrepTemplate类的典型用法代码示例。如果您正苦于以下问题:Python PrepTemplate类的具体用法?Python PrepTemplate怎么用?Python PrepTemplate使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了PrepTemplate类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_dataframe_from_template
def test_dataframe_from_template(self):
template = PrepTemplate(1)
obs = template.to_dataframe()
# 27 samples
self.assertEqual(len(obs), 27)
self.assertTrue(set(obs.index), {
u'SKB1.640202', u'SKB2.640194', u'SKB3.640195', u'SKB4.640189',
u'SKB5.640181', u'SKB6.640176', u'SKB7.640196', u'SKB8.640193',
u'SKB9.640200', u'SKD1.640179', u'SKD2.640178', u'SKD3.640198',
u'SKD4.640185', u'SKD5.640186', u'SKD6.640190', u'SKD7.640191',
u'SKD8.640184', u'SKD9.640182', u'SKM1.640183', u'SKM2.640199',
u'SKM3.640197', u'SKM4.640180', u'SKM5.640177', u'SKM6.640187',
u'SKM7.640188', u'SKM8.640201', u'SKM9.640192'})
self.assertTrue(set(obs.columns), {
u'tot_org_carb', u'common_name', u'has_extracted_data',
u'required_sample_info_status', u'water_content_soil',
u'env_feature', u'assigned_from_geo', u'altitude', u'env_biome',
u'texture', u'has_physical_specimen', u'description_duplicate',
u'physical_location', u'latitude', u'ph', u'host_taxid',
u'elevation', u'description', u'collection_timestamp',
u'taxon_id', u'samp_salinity', u'host_subject_id', u'sample_type',
u'season_environment', u'temp', u'country', u'longitude',
u'tot_nitro', u'depth', u'anonymized_name', u'target_subfragment',
u'sample_center', u'samp_size', u'run_date', u'experiment_center',
u'pcr_primers', u'center_name', u'barcodesequence', u'run_center',
u'run_prefix', u'library_construction_protocol', u'emp_status',
u'linkerprimersequence', u'experiment_design_description',
u'target_gene', u'center_project_name', u'illumina_technology',
u'sequencing_meth', u'platform', u'experiment_title',
u'study_center'})
开发者ID:jenwei,项目名称:qiita,代码行数:32,代码来源:test_util.py
示例2: add_raw_data
def add_raw_data(self, study, user, callback):
"""Adds an existing raw data to the study
Parameters
----------
study : Study
The current study object
user : User
The current user object
callback : function
The callback function to call with the results once the processing
is done
"""
msg = "Raw data successfully added"
msg_level = "success"
# Get the arguments to add the raw data
pt_id = self.get_argument('prep_template_id')
raw_data_id = self.get_argument('raw_data_id')
prep_template = PrepTemplate(pt_id)
raw_data = RawData(raw_data_id)
try:
prep_template.raw_data = raw_data
except QiitaDBError as e:
msg = html_error_message % ("adding the raw data",
str(raw_data_id), str(e))
msg = convert_text_html(msg)
callback((msg, msg_level, 'prep_template_tab', pt_id, None))
开发者ID:adamrp,项目名称:qiita,代码行数:31,代码来源:description_handlers.py
示例3: setUp
def setUp(self):
fd, self.seqs_fp = mkstemp(suffix='_seqs.fastq')
close(fd)
fd, self.barcodes_fp = mkstemp(suffix='_barcodes.fastq')
close(fd)
self.filetype = 2
self.filepaths = [(self.seqs_fp, 1), (self.barcodes_fp, 2)]
_, self.db_test_raw_dir = get_mountpoint('raw_data')[0]
with open(self.seqs_fp, "w") as f:
f.write("\n")
with open(self.barcodes_fp, "w") as f:
f.write("\n")
self._clean_up_files = []
# Create some new PrepTemplates
metadata_dict = {
'SKB8.640193': {'center_name': 'ANL',
'primer': 'GTGCCAGCMGCCGCGGTAA',
'barcode': 'GTCCGCAAGTTA',
'run_prefix': "s_G1_L001_sequences",
'platform': 'ILLUMINA',
'library_construction_protocol': 'AAAA',
'experiment_design_description': 'BBBB'}}
metadata = pd.DataFrame.from_dict(metadata_dict, orient='index')
self.pt1 = PrepTemplate.create(metadata, Study(1), "16S")
self.pt2 = PrepTemplate.create(metadata, Study(1), "18S")
self.prep_templates = [self.pt1, self.pt2]
开发者ID:MarkBruns,项目名称:qiita,代码行数:28,代码来源:test_data.py
示例4: delete_prep_template
def delete_prep_template(self, study, user, callback):
"""Delete the selected prep template
Parameters
----------
study : Study
The current study object
user : User
The current user object
callback : function
The callback function to call with the results once the processing
is done
"""
prep_template_id = int(self.get_argument('prep_template_id'))
prep_id = PrepTemplate(prep_template_id).raw_data
try:
PrepTemplate.delete(prep_template_id)
msg = ("Prep template %d has been deleted" % prep_template_id)
msg_level = "success"
prep_id = None
except Exception as e:
msg = ("Couldn't remove prep template: %s" % str(e))
msg_level = "danger"
callback((msg, msg_level, 'prep_template_tab', prep_id, None))
开发者ID:adamrp,项目名称:qiita,代码行数:26,代码来源:description_handlers.py
示例5: test_get_qiime_minimal_mapping_single_reverse_primer
def test_get_qiime_minimal_mapping_single_reverse_primer(self):
conn_handler = SQLConnectionHandler()
conn_handler
sql = """INSERT INTO qiita.prep_columns
(prep_template_id, column_name, column_type)
VALUES (1, 'reverselinkerprimer', 'varchar');
ALTER TABLE qiita.prep_1
ADD COLUMN reverselinkerprimer varchar;
DELETE FROM qiita.prep_columns
WHERE prep_template_id = 1 AND column_name = 'run_prefix';
ALTER TABLE qiita.prep_1 DROP COLUMN run_prefix;
UPDATE qiita.prep_1 SET reverselinkerprimer = %s
"""
conn_handler.execute(sql, ('GTGCCAGCM',))
prep_template = PrepTemplate(1)
prep_template.generate_files()
out_dir = mkdtemp()
obs_fps = _get_qiime_minimal_mapping(prep_template, out_dir)
exp_fps = [join(out_dir, 'prep_1_MMF.txt')]
# Check that the returned list is as expected
self.assertEqual(obs_fps, exp_fps)
# Check that the file exists
self.assertTrue(exists(exp_fps[0]))
# Check the contents of the file
with open(exp_fps[0], "U") as f:
self.assertEqual(f.read(), EXP_PREP_RLP)
开发者ID:MarkBruns,项目名称:qiita,代码行数:28,代码来源:test_processing_pipeline.py
示例6: update_prep_template
def update_prep_template(self, study, user, callback):
"""Update a prep template from the POST method
Parameters
----------
study : Study
The current study object
user : User
The current user object
callback : function
The callback function to call with the results once the processing
is done
Raises
------
HTTPError
If the prep template file does not exists
"""
# If we are on this function, the arguments "prep_template_id",
# "update_prep_template_file" must defined. If not, let tornado
# raise its error
pt_id = int(self.get_argument('prep_template_id'))
prep_template = self.get_argument('update_prep_template_file')
# Define here the message and message level in case of success
msg = "The prep template '%s' has been updated" % prep_template
msg_level = "success"
# Get the uploads folder
_, base_fp = get_mountpoint("uploads")[0]
# Get the path of the prep template in the uploads folder
fp = join(base_fp, str(study.id), prep_template)
if not exists(fp):
# The file does not exist, fail nicely
# Using 400 because we want the user to get the error in the GUI
raise HTTPError(400, "This file doesn't exist: %s" % fp)
try:
with warnings.catch_warnings(record=True) as warns:
pt = PrepTemplate(pt_id)
pt.update(load_template_to_dataframe(fp))
remove(fp)
# join all the warning messages into one. Note that this info
# will be ignored if an exception is raised
if warns:
msg = '; '.join([str(w.message) for w in warns])
msg_level = 'warning'
except (TypeError, QiitaDBColumnError, QiitaDBExecutionError,
QiitaDBDuplicateError, IOError, ValueError, KeyError,
CParserError, QiitaDBDuplicateHeaderError, QiitaDBError) as e:
# Some error occurred while processing the sample template
# Show the error to the user so they can fix the template
msg = html_error_message % ('updating the prep template:',
basename(fp), str(e))
msg = convert_text_html(msg)
msg_level = "danger"
callback((msg, msg_level, 'prep_template_tab', pt_id, None))
开发者ID:MarkBruns,项目名称:qiita,代码行数:59,代码来源:description_handlers.py
示例7: submit_VAMPS
def submit_VAMPS(preprocessed_data_id):
"""Submit preprocessed data to VAMPS
Parameters
----------
preprocessed_data_id : int
The preprocesssed data id
"""
preprocessed_data = PreprocessedData(preprocessed_data_id)
study = Study(preprocessed_data.study)
sample_template = SampleTemplate(study.sample_template)
prep_template = PrepTemplate(preprocessed_data.prep_template)
status = preprocessed_data.submitted_to_vamps_status()
if status in ('submitting', 'success'):
raise ValueError("Cannot resubmit! Current status is: %s" % status)
preprocessed_data.update_vamps_status('submitting')
# Generating a tgz
targz_folder = mkdtemp(prefix=qiita_config.working_dir)
targz_fp = join(targz_folder, '%d_%d_%d.tgz' % (study.id,
prep_template.id,
preprocessed_data.id))
targz = taropen(targz_fp, mode='w:gz')
# adding sample/prep
samp_fp = join(targz_folder, 'sample_metadata.txt')
sample_template.to_file(samp_fp)
targz.add(samp_fp, arcname='sample_metadata.txt')
prep_fp = join(targz_folder, 'prep_metadata.txt')
prep_template.to_file(prep_fp)
targz.add(prep_fp, arcname='prep_metadata.txt')
# adding preprocessed data
for _, fp, fp_type in preprocessed_data.get_filepaths():
if fp_type == 'preprocessed_fasta':
targz.add(fp, arcname='preprocessed_fasta.fna')
targz.close()
# submitting
cmd = ("curl -F user=%s -F pass='%s' -F [email protected]%s -F "
"press=UploadFile %s" % (qiita_config.vamps_user,
qiita_config.vamps_pass,
targz_fp,
qiita_config.vamps_url))
obs, _, _ = system_call(cmd)
exp = ("<html>\n<head>\n<title>Process Uploaded File</title>\n</head>\n"
"<body>\n</body>\n</html>")
if obs != exp:
preprocessed_data.update_vamps_status('failure')
return False
else:
preprocessed_data.update_vamps_status('success')
return True
开发者ID:jenwei,项目名称:qiita,代码行数:58,代码来源:commands.py
示例8: test_get_sample_names_by_run_prefix
def test_get_sample_names_by_run_prefix(self):
metadata_dict = {
'SKB8.640193': {'run_prefix': "s1", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'},
'SKD8.640184': {'run_prefix': "s2", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'},
'SKB7.640196': {'run_prefix': "s3", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'}}
md_template = pd.DataFrame.from_dict(metadata_dict, orient='index')
prep_template = PrepTemplate.create(md_template, Study(1), '16S')
for _, fp in prep_template.get_filepaths():
self.files_to_remove.append(fp)
obs = _get_sample_names_by_run_prefix(prep_template)
exp = {'s3': '1.SKB7.640196', 's2': '1.SKD8.640184',
's1': '1.SKB8.640193'}
self.assertEqual(obs, exp)
# This should raise an error
metadata_dict = {
'SKB8.640193': {'run_prefix': "s1", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'},
'SKD8.640184': {'run_prefix': "s1", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'},
'SKB7.640196': {'run_prefix': "s3", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'instrument_model': 'Illumina MiSeq',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'}}
md_template = pd.DataFrame.from_dict(metadata_dict, orient='index')
prep_template = PrepTemplate.create(md_template, Study(1), '16S')
for _, fp in prep_template.get_filepaths():
self.files_to_remove.append(fp)
with self.assertRaises(ValueError):
_get_sample_names_by_run_prefix(prep_template)
开发者ID:jenwei,项目名称:qiita,代码行数:58,代码来源:test_processing_pipeline.py
示例9: remove_add_prep_template
def remove_add_prep_template(self, fp_rpt, raw_data_id, study,
data_type_id, investigation_type, callback):
"""add prep templates
"""
PrepTemplate.create(load_template_to_dataframe(fp_rpt),
RawData(raw_data_id), study, int(data_type_id),
investigation_type=investigation_type)
remove(fp_rpt)
callback()
开发者ID:gustabf,项目名称:qiita,代码行数:10,代码来源:study_handlers.py
示例10: remove_add_study_template
def remove_add_study_template(self, raw_data, study_id, fp_rsp):
"""Replace prep templates, raw data, and sample template with a new one
"""
for rd in raw_data():
rd = RawData(rd)
for pt in rd.prep_templates:
if PrepTemplate.exists(pt):
PrepTemplate.delete(pt)
if SampleTemplate.exists(study_id):
SampleTemplate.delete(study_id)
SampleTemplate.create(load_template_to_dataframe(fp_rsp),
Study(study_id))
remove(fp_rsp)
开发者ID:RNAer,项目名称:qiita,代码行数:14,代码来源:description_handlers.py
示例11: update_investigation_type
def update_investigation_type(self, study, user, callback):
"""Updates the investigation type of a prep template
Parameters
----------
study : Study
The current study object
user : User
The current user object
callback : function
The callback function to call with the results once the processing
is done
"""
msg = "investigation type successfully updated"
msg_level = "success"
ppd_id = int(self.get_argument('ppd_id'))
prep_id = self.get_argument('prep_id')
edit_investigation_type = self.get_argument('edit-investigation-type',
None)
edit_user_defined_investigation_type = self.get_argument(
'edit-user-defined-investigation-type', None)
edit_new_investigation_type = self.get_argument(
'edit-new-investigation-type', None)
pt = PrepTemplate(prep_id)
investigation_type = self._process_investigation_type(
edit_investigation_type, edit_user_defined_investigation_type,
edit_new_investigation_type)
try:
pt.investigation_type = investigation_type
except QiitaDBColumnError as e:
msg = html_error_message % (", invalid investigation type: ",
investigation_type, str(e))
msg = convert_text_html(msg)
msg_level = "danger"
if ppd_id == 0:
top_tab = "prep_template_tab"
sub_tab = prep_id
prep_tab = None
else:
top_tab = "preprocessed_data_tab"
sub_tab = ppd_id
prep_tab = None
callback((msg, msg_level, top_tab, sub_tab, prep_tab))
开发者ID:adamrp,项目名称:qiita,代码行数:50,代码来源:description_handlers.py
示例12: preprocessor
def preprocessor(study_id, prep_template_id, param_id, param_constructor):
"""Dispatch for preprocessor work"""
study = Study(study_id)
prep_template = PrepTemplate(prep_template_id)
params = param_constructor(param_id)
sp = StudyPreprocessor()
try:
preprocess_out = sp(study, prep_template, params)
except Exception as e:
error_msg = ''.join(format_exception_only(e, exc_info()))
prep_template.preprocessing_status = "failed: %s" % error_msg
preprocess_out = None
return preprocess_out
开发者ID:jwdebelius,项目名称:qiita,代码行数:15,代码来源:dispatchable.py
示例13: remove_add_prep_template
def remove_add_prep_template(self, fp_rpt, study, data_type_id, investigation_type):
"""add prep templates"""
pt_id = PrepTemplate.create(
load_template_to_dataframe(fp_rpt), study, _to_int(data_type_id), investigation_type=investigation_type
).id
remove(fp_rpt)
return pt_id
开发者ID:DarcyMyers,项目名称:qiita,代码行数:7,代码来源:description_handlers.py
示例14: test_create
def test_create(self):
"""Creates a new PrepTemplate"""
pt = PrepTemplate.create(self.metadata, self.new_raw_data)
# The returned object has the correct id
self.assertEqual(pt.id, 3)
# The relevant rows to common_prep_info have been added.
obs = self.conn_handler.execute_fetchall(
"SELECT * FROM qiita.common_prep_info WHERE raw_data_id=3")
# raw_data_id, sample_id, center_name, center_project_name,
# ebi_submission_accession, ebi_study_accession, emp_status_id,
# data_type_id
exp = [[3, 'SKB8.640193', 'ANL', 'Test Project', None, None, 1, 2],
[3, 'SKD8.640184', 'ANL', 'Test Project', None, None, 1, 2],
[3, 'SKB7.640196', 'ANL', 'Test Project', None, None, 1, 2]]
self.assertEqual(sorted(obs), sorted(exp))
# The relevant rows have been added to the raw_data_prep_columns
obs = self.conn_handler.execute_fetchall(
"SELECT * FROM qiita.raw_data_prep_columns WHERE raw_data_id=3")
# raw_data_id, column_name, column_type
exp = [[3, "str_column", "varchar"]]
self.assertEqual(obs, exp)
# The new table exists
self.assertTrue(exists_table("prep_3", self.conn_handler))
# The new table hosts the correct values
obs = self.conn_handler.execute_fetchall(
"SELECT * FROM qiita.prep_3")
# sample_id, str_column
exp = [['SKB8.640193', "Value for sample 1"],
['SKD8.640184', "Value for sample 2"],
['SKB7.640196', "Value for sample 3"]]
self.assertEqual(sorted(obs), sorted(exp))
开发者ID:teravest,项目名称:qiita,代码行数:35,代码来源:test_metadata_template.py
示例15: test_get_preprocess_fastq_cmd_per_sample_FASTQ_failure
def test_get_preprocess_fastq_cmd_per_sample_FASTQ_failure(self):
metadata_dict = {
'SKB8.640193': {'run_prefix': "sample1_failure", 'primer': 'A',
'barcode': 'A', 'center_name': 'ANL',
'platform': 'ILLUMINA',
'library_construction_protocol': 'A',
'experiment_design_description': 'A'}}
md_template = pd.DataFrame.from_dict(metadata_dict, orient='index')
prep_template = PrepTemplate.create(md_template, Study(1), '16S')
# This part should fail
fp1 = self.path_builder('sample1_failure.fastq')
with open(fp1, 'w') as f:
f.write('\n')
self.files_to_remove.append(fp1)
fp2 = self.path_builder('sample1_failure.barcodes.fastq.gz')
with open(fp2, 'w') as f:
f.write('\n')
self.files_to_remove.append(fp2)
forward_filepath_id = convert_to_id('raw_forward_seqs',
'filepath_type')
barcode_filepath_id = convert_to_id('raw_barcodes', 'filepath_type')
fps = [(fp1, forward_filepath_id), (fp2, barcode_filepath_id)]
filetype_id = get_filetypes()['per_sample_FASTQ']
raw_data = RawData.create(filetype_id, [prep_template], fps)
params = [p for p in list(PreprocessedIlluminaParams.iter())
if p.name == 'per sample FASTQ defaults'][0]
with self.assertRaises(ValueError):
_get_preprocess_fastq_cmd(raw_data, prep_template, params)
开发者ID:MarkBruns,项目名称:qiita,代码行数:32,代码来源:test_processing_pipeline.py
示例16: test_get_qiime_minimal_mapping_multiple
def test_get_qiime_minimal_mapping_multiple(self):
# We need to create a prep template in which we have different run
# prefix values, so we can test this case
metadata_dict = {
'SKB8.640193': {'center_name': 'ANL',
'center_project_name': 'Test Project',
'ebi_submission_accession': None,
'EMP_status': 'EMP',
'str_column': 'Value for sample 1',
'primer': 'GTGCCAGCMGCCGCGGTAA',
'barcode': 'GTCCGCAAGTTA',
'run_prefix': "s_G1_L001_sequences",
'platform': 'ILLUMINA',
'library_construction_protocol': 'AAA',
'experiment_design_description': 'BBB'},
'SKD8.640184': {'center_name': 'ANL',
'center_project_name': 'Test Project',
'ebi_submission_accession': None,
'EMP_status': 'EMP',
'str_column': 'Value for sample 2',
'primer': 'GTGCCAGCMGCCGCGGTAA',
'barcode': 'CGTAGAGCTCTC',
'run_prefix': "s_G1_L001_sequences",
'platform': 'ILLUMINA',
'library_construction_protocol': 'AAA',
'experiment_design_description': 'BBB'},
'SKB7.640196': {'center_name': 'ANL',
'center_project_name': 'Test Project',
'ebi_submission_accession': None,
'EMP_status': 'EMP',
'str_column': 'Value for sample 3',
'primer': 'GTGCCAGCMGCCGCGGTAA',
'barcode': 'CCTCTGAGAGCT',
'run_prefix': "s_G1_L002_sequences",
'platform': 'ILLUMINA',
'library_construction_protocol': 'AAA',
'experiment_design_description': 'BBB'}
}
md_template = pd.DataFrame.from_dict(metadata_dict, orient='index')
prep_template = PrepTemplate.create(md_template, Study(1), '16S')
for _, fp in prep_template.get_filepaths():
self.files_to_remove.append(fp)
out_dir = mkdtemp()
obs_fps = sorted(_get_qiime_minimal_mapping(prep_template, out_dir))
exp_fps = sorted([join(out_dir, 's_G1_L001_sequences_MMF.txt'),
join(out_dir, 's_G1_L002_sequences_MMF.txt')])
# Check that the returned list is as expected
self.assertEqual(obs_fps, exp_fps)
# Check that the file exists
for fp in exp_fps:
self.assertTrue(exists(fp))
# Check the contents of the file
for fp, contents in zip(exp_fps, [EXP_PREP_1, EXP_PREP_2]):
with open(fp, "U") as f:
self.assertEqual(f.read(), contents)
开发者ID:MarkBruns,项目名称:qiita,代码行数:58,代码来源:test_processing_pipeline.py
示例17: get_prep_templates
def get_prep_templates(self, raw_data, callback):
"""Get all prep templates for a list of raw data objects"""
d = {}
for rd in raw_data:
# We neeed this so PrepTemplate(p) doesn't fail if that raw
# doesn't exist but raw data has the row: #554
prep_templates = sorted(rd.prep_templates)
d[rd.id] = [PrepTemplate(p) for p in prep_templates
if PrepTemplate.exists(p)]
callback(d)
开发者ID:BrindhaBioinfo,项目名称:qiita,代码行数:10,代码来源:description_handlers.py
示例18: test_to_file
def test_to_file(self):
"""to file writes a tab delimited file with all the metadata"""
fd, fp = mkstemp()
close(fd)
pt = PrepTemplate.create(self.metadata, self.new_raw_data)
pt.to_file(fp)
self._clean_up_files.append(fp)
with open(fp, 'U') as f:
obs = f.read()
self.assertEqual(obs, EXP_PREP_TEMPLATE)
开发者ID:teravest,项目名称:qiita,代码行数:10,代码来源:test_metadata_template.py
示例19: remove_add_study_template
def remove_add_study_template(self, raw_data, study_id, fp_rsp, data_type, is_mapping_file):
"""Replace prep templates, raw data, and sample template with a new one
"""
if is_mapping_file and data_type == "":
raise ValueError("Please, choose a data type if uploading a QIIME " "mapping file")
for rd in raw_data():
rd = RawData(rd)
for pt in rd.prep_templates:
if PrepTemplate.exists(pt):
PrepTemplate.delete(pt)
if SampleTemplate.exists(study_id):
SampleTemplate.delete(study_id)
if is_mapping_file:
create_templates_from_qiime_mapping_file(fp_rsp, Study(study_id), int(data_type))
else:
SampleTemplate.create(load_template_to_dataframe(fp_rsp), Study(study_id))
remove(fp_rsp)
开发者ID:DarcyMyers,项目名称:qiita,代码行数:20,代码来源:description_handlers.py
示例20: test_get_qiime_minimal_mapping_single_no_run_prefix
def test_get_qiime_minimal_mapping_single_no_run_prefix(self):
conn_handler = SQLConnectionHandler()
sql = """DELETE FROM qiita.prep_columns
WHERE prep_template_id = 1 AND column_name = 'run_prefix';
ALTER TABLE qiita.prep_1 DROP COLUMN run_prefix"""
conn_handler.execute(sql)
prep_template = PrepTemplate(1)
prep_template.generate_files()
out_dir = mkdtemp()
obs_fps = _get_qiime_minimal_mapping(prep_template, out_dir)
exp_fps = [join(out_dir, 'prep_1_MMF.txt')]
# Check that the returned list is as expected
self.assertEqual(obs_fps, exp_fps)
# Check that the file exists
self.assertTrue(exists(exp_fps[0]))
# Check the contents of the file
with open(exp_fps[0], "U") as f:
self.assertEqual(f.read(), EXP_PREP)
开发者ID:MarkBruns,项目名称:qiita,代码行数:20,代码来源:test_processing_pipeline.py
注:本文中的qiita_db.metadata_template.PrepTemplate类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论