Initial setup for element aviti

.devcontainer/devcontainer.json

@@ -12,7 +12,12 @@
   "features": {},
   "customizations": {
     "vscode": {
-      "extensions": ["ms-python.python", "eamodio.gitlens"]
+      "extensions": [
+        "ms-python.python",
+        "eamodio.gitlens",
+        "charliermarsh.ruff",
+        "ms-python.mypy-type-checker"
+      ]
     }
   },
   // Features to add to the dev container. More info: https://containers.dev/features.

Dockerfile

@@ -23,7 +23,6 @@ COPY requirements-dev.txt requirements-dev.txt
 RUN python -m pip install -r requirements-dev.txt
 
 RUN mkdir /root/.taca/
-COPY tests/data/taca_test_cfg.yaml /root/.taca/taca.yaml
 
 FROM base AS testing
 COPY . /taca

VERSIONLOG.md

@@ -1,5 +1,17 @@
 # TACA Version Log
 
+## 20241016.1
+
+Fix wrong logic for collecting unassigned indexes
+
+## 20241011.1
+
+Fix issue with 0 lane number; Add percentage of unassigned in total unassigned per lane
+
+## 20241008.1
+
+Add support for processing Element Aviti data
+
 ## 20241007.1
 
 Update taca-ngi-pipeline repo URL in README

pyproject.toml

@@ -1,5 +1,3 @@
-title = "taca"
-
 # === LINTING ================================================================
 
 [tool.ruff]
@@ -37,7 +35,12 @@ filterwarnings = [
     'ignore::DeprecationWarning:couchdb.*',
     'ignore::DeprecationWarning:pkg_resources.*',
 ]
-addopts = "--cov=./taca --cov-report term-missing -vv --cache-clear tests/"
+# Default addopts
+addopts = "--ignore tests_old/"
+
+# CLI coverage reports, messes with IDE debugging
+# pytest --ignore tests_old/ --cov=./taca --cov-report term-missing -vv
+
 
 [tool.coverage.run]
 # The comment "# pragma: no cover" can be used to exclude a line from coverage

taca/__init__.py

@@ -1,3 +1,3 @@
 """Main TACA module"""
 
-__version__ = "1.0.0"
+__version__ = "1.1.0"

taca/analysis/analysis_element.py

@@ -0,0 +1,163 @@
+"""Analysis methods for sequencing runs produced by Element instruments."""
+
+import glob
+import logging
+import os
+
+from taca.element.Aviti_Runs import Aviti_Run
+from taca.utils.config import CONFIG
+from taca.utils.misc import send_mail
+
+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
+        """
+        try:
+            run.parse_run_parameters()
+        except FileNotFoundError:
+            logger.warning(
+                f"Cannot reliably set NGI_run_id for {run} due to missing RunParameters.json. Aborting run processing"
+            )
+            email_subject = f"Issues processing {run}"
+            email_message = (
+                f"RunParameters.json missing for {run}. Processing was aborted."
+            )
+            send_mail(email_subject, email_message, CONFIG["mail"]["recipients"])
+            raise
+
+        #### Sequencing status ####
+        sequencing_done = run.check_sequencing_status()
+        if not sequencing_done:
+            run.status = "sequencing"
+            if run.status_changed:
+                run.update_statusdb()
+            return
+
+        #### Demultiplexing status ####
+        demultiplexing_status = run.get_demultiplexing_status()
+        if demultiplexing_status == "not started":
+            lims_zip_path = run.find_lims_zip()
+            if lims_zip_path is not None:
+                os.mkdir(run.demux_dir)
+                run.copy_manifests(lims_zip_path)
+                demux_manifests = run.make_demux_manifests(
+                    manifest_to_split=run.lims_manifest
+                )
+                sub_demux_count = 0
+                for demux_manifest in sorted(demux_manifests):
+                    sub_demux_dir = os.path.join(
+                        run.run_dir, f"Demultiplexing_{sub_demux_count}"
+                    )
+                    os.mkdir(sub_demux_dir)
+                    run.start_demux(demux_manifest, sub_demux_dir)
+                    sub_demux_count += 1
+                run.status = "demultiplexing"
+                if run.status_changed():
+                    run.update_statusdb()
+                return
+            else:
+                logger.warning(
+                    f"Run manifest is missing for {run}, demultiplexing aborted"
+                )
+                email_subject = f"Issues processing {run}"
+                email_message = (
+                    f"Run manifest is missing for {run}, demultiplexing aborted"
+                )
+                send_mail(email_subject, email_message, CONFIG["mail"]["recipients"])
+                return
+        elif demultiplexing_status == "ongoing":
+            run.status = "demultiplexing"
+            if run.status_changed:
+                run.update_statusdb()
+            return
+
+        elif demultiplexing_status != "finished":
+            logger.warning(
+                f"Unknown demultiplexing status {demultiplexing_status} of run {run}. Please investigate."
+            )
+            email_subject = f"Issues processing {run}"
+            email_message = f"Unknown demultiplexing status {demultiplexing_status} of run {run}. Please investigate."
+            send_mail(email_subject, email_message, CONFIG["mail"]["recipients"])
+            return
+
+        #### Transfer status ####
+        transfer_status = run.get_transfer_status()
+        if transfer_status == "not started":
+            demux_results_dirs = glob.glob(
+                os.path.join(run.run_dir, "Demultiplexing_*")
+            )
+            run.aggregate_demux_results(demux_results_dirs)
+            run.sync_metadata()
+            run.make_transfer_indicator()
+            run.status = "transferring"
+            if run.status_changed:
+                run.update_statusdb()
+                # TODO: Also update statusdb with a timestamp of when the transfer started
+            run.transfer()
+            return
+        elif transfer_status == "ongoing":
+            run.status = "transferring"
+            if run.status_changed:
+                run.update_statusdb()
+            logger.info(
+                f"{run} is being transferred. Skipping."
+            )  # TODO: fix formatting, currently prints "ElementRun(20240910_AV242106_B2403418431) is being transferred"
+            return
+        elif transfer_status == "rsync done":
+            if run.rsync_successful():
+                run.remove_transfer_indicator()
+                run.update_transfer_log()
+                run.status = "transferred"
+                if run.status_changed:
+                    run.update_statusdb()
+                run.archive()
+                run.status = "archived"
+
+                if run.status_changed:
+                    run.update_statusdb()
+            else:
+                run.status = "transfer failed"
+                logger.warning(
+                    f"An issue occurred while transfering {run} to the analysis cluster."
+                )
+                email_subject = f"Issues processing {run}"
+                email_message = f"An issue occurred while transfering {run} to the analysis cluster."
+                send_mail(email_subject, email_message, CONFIG["mail"]["recipients"])
+            return
+        else:
+            logger.warning(
+                f"Unknown transfer status {transfer_status} of run {run}, please investigate."
+            )
+            email_subject = f"Issues processing {run}"
+            email_message = f"Unknown transfer status {transfer_status} of run {run}, please investigate."
+            send_mail(email_subject, email_message, CONFIG["mail"]["recipients"])
+            return
+
+    if given_run:
+        run = Aviti_Run(given_run, CONFIG)
+        _process(run)
+    else:
+        data_dirs = CONFIG.get("element_analysis").get("data_dirs")
+        for data_dir in data_dirs:
+            # Run folder looks like DATE_*_*, the last section is the FC side (A/B) and name
+            runs = glob.glob(os.path.join(data_dir, "[1-9]*_*_*"))
+            for run in runs:
+                runObj = Aviti_Run(run, CONFIG)
+                try:
+                    _process(runObj)
+                except:
+                    # 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}")
+                    # TODO: Think about how to avoid silent errors (email?)
+                    pass

taca/analysis/cli.py

@@ -3,7 +3,7 @@
 import click
 
 from taca.analysis import analysis as an
-from taca.analysis import analysis_nanopore
+from taca.analysis import analysis_element, analysis_nanopore
 
 
 @click.group()
@@ -72,6 +72,29 @@ def updatedb(rundir, software):
     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,8 @@
+from taca.element.Element_Runs import Run
+
+
+class Aviti_Run(Run):
+    def __init__(self, run_dir, configuration):
+        self.sequencer_type = "Aviti"
+        self.demux_dir = "Demultiplexing"
+        super().__init__(run_dir, configuration)