Initial setup for element aviti

taca/analysis/analysis_element.py

@@ -0,0 +1,65 @@
+"""Analysis methods for sequencing runs produced by Element instruments."""
+
+import glob
+import logging
+import os
+
+from taca.element.Element_Runs import Aviti_Run
+from taca.utils.config import CONFIG
+
+logger = logging.getLogger(__name__)
+
+
+def run_preprocessing(given_run):
+    """Run demultiplexing in all data directories.
+
+    :param str given_run: Process a particular run instead of looking for runs
+    """
+
+    def _process(run):
+        """Process a run/flowcell and transfer to analysis server.
+
+        :param taca.element.Run run: Run to be processed and transferred
+        """
+        # Check if sequencing is finished. (is the final file there and was it completed OK)
+        # if sequencing is not done
+            # Update statusdb?
+            # return
+        # else If sequencing finished and demux not started
+            # Update statusdb
+            # Get/generate sample sheet
+            # Start demux
+        # else if sequencing finished and demux ongoing
+            # do nothing
+        # Else if sequencing started and demux finished
+            # 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
+                # transfer run to miarka 
+                # remove hidden file if transfer was successful
+                # Update transfer log
+                # archive run to nosync
+            # elif run is being transferred (hidden file exists)
+                # return
+            # elif run is already transferred (in transfer log)
+                # warn that transferred run has not been archived
+
+
+
+    if given_run:
+        run = Aviti_Run(run) #TODO: Needs to change if more Element machines are aquired in the future
+        _process(runObj)
+    else:
+        data_dirs = CONFIG.get("element_analysis").get("data_dirs") #TODO: add to config
+        for data_dir in data_dirs:
+            # Run folder looks like DATE_*_*_*, the last section is the FC name.
+            runs = glob.glob(os.path.join(data_dir, "[1-9]*_*_*_*")) #TODO: adapt to aviti format
+            for run in runs:
+                runObj = Aviti_Run(run)
+                try:
+                    _process(runObj)
+                except: #TODO: chatch error message and print it
+                    # This function might throw and exception,
+                    # it is better to continue processing other runs
+                    logger.warning(f"There was an error processing the run {run}")
+                    pass

taca/analysis/cli.py

@@ -4,6 +4,7 @@
 
 from taca.analysis import analysis as an
 from taca.analysis import analysis_nanopore
+from taca.analysis import analysis_element
 
 
 @click.group()
@@ -71,6 +72,28 @@ def updatedb(rundir, software):
     """Save the run to statusdb."""
     an.upload_to_statusdb(rundir, software)
 
+# Element analysis subcommands
+
+
+@analysis.command()
+@click.option(
+    "-r",
+    "--run",
+    type=click.Path(exists=True),
+    default=None,
+    help="Demultiplex only a particular run",
+)
+def demultiplex_element(run):
+    """Demultiplex and transfer all runs present in the data directories."""
+    analysis_element.run_preprocessing(run)
+
+
+@analysis.command()
+@click.argument("run")
+def element_updatedb(run):
+    """Save the run to statusdb."""
+    analysis_element.upload_to_statusdb(run)
+
 
 # Nanopore analysis subcommands
 

taca/element/Aviti_Runs.py

@@ -0,0 +1,7 @@
+from taca.element.Element_Runs import Run
+
+
+class Aviti_Run(Run):
+    def __init__(self, run_dir, configuration):
+        super().__init__(run_dir, configuration)
+        self.sequencer_type = "Aviti"

taca/element/Element_Runs.py

@@ -0,0 +1,18 @@
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+
+
+class Run:
+    """Defines an Element run"""
+
+    def __init__(self, run_dir, configuration):
+        if not os.path.exists(run_dir):
+            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"
+
+    def is_transferred(self, transfer_file):
+        pass

taca/element/__init__.py

@@ -0,0 +1,3 @@
+"""
+Classes to parse and work with Element data
+"""