added flex sample dir listing

igf_airflow/utils/dag46_scRNA_10X_flex_utils.py

@@ -128,6 +128,38 @@ def prepare_cellranger_flex_script(
       "run_script": run_script_file,
       "output_dir": os.path.join(work_dir, sample_group)}
     return analysis_script_info
+  except Exception as e:
+    log.error(e)
+    send_airflow_failed_logs_to_channels(
+      ms_teams_conf=MS_TEAMS_CONF,
+      message_prefix=e)
+    raise ValueError(e)
+
+
+## TASK
+@task(
+  task_id="flex_sample_factory",
+  retry_delay=timedelta(minutes=5),
+  retries=4,
+  queue='hpc_4G')
+def flex_sample_factory(
+  cellranger_output_dir: str) \
+    -> list[str]:
+  try:
+    ## set cellranger counts dir
+    cellranger_counts_dir = \
+      os.path.join(
+        cellranger_output_dir,
+        'outs',
+        'per_sample_outs')
+    check_file_path(cellranger_counts_dir)
+    ## get dirs
+    sample_list = \
+      os.listdir(cellranger_counts_dir)
+    if len(sample_list) == 0:
+      raise ValueError(
+        f"Missing flex output in path: {cellranger_counts_dir}")
+    return sample_list
   except Exception as e:
     log.error(e)
     send_airflow_failed_logs_to_channels(

test/igf_airflow/test_dag46_scRNA_10X_flex_utils.py

@@ -8,7 +8,8 @@
 from yaml import load, SafeLoader
 from igf_airflow.utils.dag46_scRNA_10X_flex_utils import (
     _get_cellranger_sample_group,
-    prepare_cellranger_flex_script)
+    prepare_cellranger_flex_script,
+    flex_sample_factory)
 from igf_data.igfdb.igfTables import Base
 from igf_data.igfdb.baseadaptor import BaseAdaptor
 from igf_data.igfdb.projectadaptor import ProjectAdaptor
@@ -258,5 +259,29 @@ def test_prepare_cellranger_flex_script(self):
         assert sample_group == "grp1"
         assert "--id=grp1" in data
 
+  def test_flex_sample_factory(self):
+    cellranger_output_dir = \
+      os.path.join(
+        self.temp_dir,
+        'cellranger_output_dir',
+        'outs',
+        'per_sample_outs')
+    os.makedirs(
+      cellranger_output_dir,
+      exist_ok=True)
+    for sample_id in ['SAMPLE1', 'SAMPLE2']:
+      os.makedirs(
+        os.path.join(
+          cellranger_output_dir,
+          sample_id),
+        exist_ok=True)
+    sample_list = \
+      flex_sample_factory.function(
+        cellranger_output_dir=os.path.join(
+          self.temp_dir,
+          'cellranger_output_dir'))
+    assert len(sample_list) == 2
+    assert "SAMPLE1" in sample_list
+
 if __name__=='__main__':
   unittest.main()