Process aviti data from sequencing to demux

taca/analysis/analysis_element.py

@@ -6,9 +6,17 @@
 
 from taca.element.Element_Runs import Aviti_Run
 from taca.utils.config import CONFIG
+from taca.utils import statusdb
+
 
 logger = logging.getLogger(__name__)
 
+def _upload_to_statusdb(run):
+    """Triggers the upload to statusdb.
+
+    :param Run run: the object run
+    """
+    pass
 
 def run_preprocessing(given_run):
     """Run demultiplexing in all data directories.
@@ -21,21 +29,26 @@ def _process(run):
 
         :param taca.element.Run run: Run to be processed and transferred
         """
-        # Fetch statusdb document for run
-
-        # Get previous status of run from statusdb document
-        # Check if sequencing is finished. (is the final file there and was it completed OK)
-        # if sequencing is not done
-            # compare previous status with current status and update statusdb document if different
-            # return
-        # else if sequencing finished and demux not started
-            # Get/generate sample sheet
+        #TODO: Fetch statusdb document for run
+        #TODO: Get previous status of run from statusdb document
+        sequencing_done = run.check_sequencing_status()
+        demultiplexing_status = run.get_demultiplexing_status()
+        if not sequencing_done:
+            #TODO: compare previous status with current status and update statusdb document if different
+            return
+        elif sequencing_done and demultiplexing_status == "not started":
+            if not run.manifest_exists(): # Assumes that we use the same manifest as for sequencing. TODO: demux settings need to be added to the original manifest by lims
+                #TODO: email operator that manifest is missing
+                return
             # Start demux
-            # compare previous status with current status and update statusdb document if different
-        # else if sequencing finished and demux ongoing
-            # compare previous status with current status and update statusdb document if different
-            # return
-        # Else if sequencing started and demux finished
+            run.start_demux()
+            #TODO: compare previous status with current status and update statusdb document if different
+            return
+        elif sequencing_done and demultiplexing_status == "ongoing":
+            #TODO: compare previous status with current status and update statusdb document if different
+            return
+        elif sequencing_done and demultiplexing_status == "finished":
+            # Sync metadata to ngi-data-ns
             # check if run is transferred or transfer is ongoing
             # if run has not been transferred and transfer is not ongoing
                 # make a hidden file to indicate that transfer has started
@@ -53,7 +66,7 @@ def _process(run):
             # elif run is already transferred (in transfer log)
                 # compare previous status with current status and update statusdb document if different
                 # warn that transferred run has not been archived
-
+            pass
 
 
     if given_run:

taca/element/Element_Runs.py

@@ -1,5 +1,10 @@
 import logging
 import os
+import json
+from datetime import datetime
+
+from taca.utils import misc
+from taca.utils.filesystem import chdir
 
 logger = logging.getLogger(__name__)
 
@@ -12,7 +17,51 @@ def __init__(self, run_dir, configuration):
             raise RuntimeError(f"Could not locate run directory {run_dir}")
         self.run_dir = os.path.abspath(run_dir)
         self.CONFIG = configuration
-        self.demux_dir = "Demultiplexing"
+        self.demux_dir = os.path.join(self.run_dir, "Demultiplexing")
+        self.final_sequencing_file = os.path.join(self.run_dir, "RunUploaded.json")
+        self.demux_stats_file = os.path.join(self.demux_dir, "RunStats.json") #TODO: How to handle SideA/SideB?
+        self.run_manifest_file = os.path.join(self.run_dir, "RunManifest.csv")
+
+    def check_sequencing_status(self):
+        if os.path.exists(self.final_sequencing_file):
+            with open(self.final_sequencing_file) as json_file:
+                sequencing_outcome = json.load(json_file).get("outcome")
+            if sequencing_outcome != "OutcomeCompleted":
+                return False
+            else:
+                return True
+        else:
+            return False
+
+    def get_demultiplexing_status(self):
+        if not os.path.exists(self.demux_dir):
+            return "not started"
+        elif os.path.exists(self.demux_dir) and not os.path.isfile(self.demux_stats_file):
+            return "ongoing"
+        elif os.path.exists(self.demux_dir) and os.path.isfile(self.demux_stats_file):
+            return "finished"
+
+    def manifest_exists(self):
+        return os.path.isfile(self.run_manifest_file)
+
+    def generate_demux_command(self):
+        command = [self.CONFIG.get(self.software)["bin"], #TODO add path to bases2fastq executable to config
+                   self.run_dir,
+                   self.demux_dir, #TODO: how to handle SideA/SideB?
+                   "-p 12"
+                   ] 
+        return command
+
+    def start_demux(self):
+        with chdir(self.run_dir):
+            cmd = self.generate_demux_command()
+            misc.call_external_command_detached(
+                cmd, with_log_files=True, prefix=f"demux_"
+            )
+            logger.info(
+                "Bases2Fastq conversion and demultiplexing "
+                f"started for run {os.path.basename(self.id)} on {datetime.now()}"
+            )
 
     def is_transferred(self, transfer_file):
         pass