commit before changing branches

q2_amr/card/reads.py

@@ -74,13 +74,24 @@ def annotate_reads_card(
                 path=path_allele, samp_bin_name=samp, data_type="reads"
             )
             allele_frequency_list.append(allele_frequency)
+
             path_gene = os.path.join(samp_input_dir, "output.gene_mapping_data.txt")
             gene_frequency = read_in_txt(
                 path=path_gene, samp_bin_name=samp, data_type="reads"
             )
             gene_frequency_list.append(gene_frequency)
-            move_files(samp_input_dir, samp_allele_dir, "allele")
-            move_files(samp_input_dir, samp_gene_dir, "gene")
+
+            for map_type, des_dir in zip(
+                ["allele", "gene"], [samp_allele_dir, samp_gene_dir]
+            ):
+                files = [f"{map_type}_mapping_data.txt", "overall_mapping_stats.txt"]
+                files.append("sorted.length_100.bam") if map_type == "allele" else None
+
+                for file in files:
+                    shutil.copy(
+                        os.path.join(samp_input_dir, "output." + file),
+                        os.path.join(des_dir, file),
+                    )
 
     allele_feature_table = create_count_table(allele_frequency_list)
     gene_feature_table = create_count_table(gene_frequency_list)
@@ -92,21 +103,6 @@ def annotate_reads_card(
     )
 
 
-def move_files(source_dir: str, des_dir: str, map_type: str):
-    shutil.move(
-        os.path.join(source_dir, f"output.{map_type}_mapping_data.txt"),
-        os.path.join(des_dir, f"{map_type}_mapping_data.txt"),
-    )
-    shutil.copy(
-        os.path.join(source_dir, "output.overall_mapping_stats.txt"),
-        os.path.join(des_dir, "overall_mapping_stats.txt"),
-    )
-    shutil.copy(
-        os.path.join(source_dir, "output.sorted.length_100.bam"),
-        os.path.join(des_dir, "sorted.length_100.bam"),
-    )
-
-
 def run_rgi_bwt(
     cwd: str,
     samp: str,

q2_amr/tests/card/test_reads.py

@@ -13,7 +13,6 @@
 from q2_amr.card.reads import (
     annotate_reads_card,
     extract_sample_stats,
-    move_files,
     plot_sample_stats,
     run_rgi_bwt,
     visualize_annotation_stats,
@@ -146,13 +145,13 @@ def annotate_reads_card_test_body(self, read_type):
 
             # Assert if the expected files are in every sample directory and in both
             # resulting CARD annotation objects
+
             for num in [0, 1]:
                 map_type = "allele" if num == 0 else "gene"
+                files = [f"{map_type}_mapping_data.txt", "overall_mapping_stats.txt"]
+                files.append("sorted.length_100.bam") if map_type == "allele" else None
                 for samp in ["sample1", "sample2"]:
-                    for file in [
-                        f"{map_type}_mapping_data.txt",
-                        "overall_mapping_stats.txt",
-                    ]:
+                    for file in files:
                         self.assertTrue(
                             os.path.exists(os.path.join(str(result[num]), samp, file))
                         )
@@ -205,36 +204,6 @@ def test_exception_raised(self):
             run_rgi_bwt()
             self.assertEqual(str(cm.exception), expected_message)
 
-    def test_move_files_allele(self):
-        self.move_files_test_body("allele")
-
-    def test_move_files_gene(self):
-        self.move_files_test_body("gene")
-
-    def move_files_test_body(self, map_type):
-        with tempfile.TemporaryDirectory() as tmp:
-            source_dir = os.path.join(tmp, "source_dir")
-            des_dir = os.path.join(tmp, "des_dir")
-
-            os.mkdir(source_dir)
-            os.mkdir(des_dir)
-
-            files = [
-                f"output.{map_type}_mapping_data.txt",
-                "output.overall_mapping_stats.txt",
-                "output.sorted.length_100.bam",
-            ]
-
-            for file in files:
-                file_path = self.get_data_path(file)
-                shutil.copy(file_path, source_dir)
-
-            move_files(source_dir, des_dir, map_type)
-
-            for file in files:
-                file_path = os.path.join(des_dir, file[7:])
-                self.assertTrue(os.path.exists(file_path))
-
     def test_extract_sample_stats(self):
         with tempfile.TemporaryDirectory() as tmp:
             mapping_stats_path = self.get_data_path("output.overall_mapping_stats.txt")

q2_amr/types/_format.py

@@ -415,12 +415,14 @@ class CARDAlleleAnnotationDirectoryFormat(
     MultiDirValidationMixin, model.DirectoryFormat
 ):
     allele = model.FileCollection(
-        "allele_mapping_data.txt", format=CARDAlleleAnnotationFormat
+        r".+allele_mapping_data.txt$", format=CARDAlleleAnnotationFormat
     )
     stats = model.FileCollection(
-        "overall_mapping_stats.txt", format=CARDAnnotationStatsFormat
+        r".+overall_mapping_stats.txt$", format=CARDAnnotationStatsFormat
+    )
+    bam = model.FileCollection(
+        r".+sorted.length_100.bam$", format=CARDAnnotationBamFormat
     )
-    bam = model.FileCollection("sorted.length_100.bam", format=CARDAnnotationBamFormat)
 
     @allele.set_path_maker
     def allele_path_maker(self, sample_id):
@@ -437,10 +439,10 @@ def bam_path_maker(self, sample_id):
 
 class CARDGeneAnnotationDirectoryFormat(MultiDirValidationMixin, model.DirectoryFormat):
     gene = model.FileCollection(
-        r".+(gene_mapping_data.txt)$", format=CARDGeneAnnotationFormat
+        r".+gene_mapping_data.txt$", format=CARDGeneAnnotationFormat
     )
     stats = model.FileCollection(
-        r".+(overall_mapping_stats.txt)$", format=CARDAnnotationStatsFormat
+        r".+overall_mapping_stats.txt$", format=CARDAnnotationStatsFormat
     )
 
     @gene.set_path_maker