Merge pull request #1159 from fls-bioinformatics-core/abstract-updating-functionality

Consolidate path updating functionality in 'archive', 'clone' and 'update' commands

Author: pjbriggs

auto_process_ngs/auto_processor.py

@@ -10,13 +10,9 @@
 
 import sys
 import os
-import subprocess
 import logging
 import shutil
-import uuid
 import time
-import ast
-import gzip
 import atexit
 import bcftbx.IlluminaData as IlluminaData
 import bcftbx.utils as bcf_utils
@@ -25,6 +21,7 @@
 from .analysis import run_reference_id
 from .metadata import AnalysisDirParameters
 from .metadata import AnalysisDirMetadata
+from .metadata import AnalysisProjectInfo
 from .metadata import ProjectMetadataFile
 from .utils import edit_file
 from .utils import get_numbered_subdir
@@ -140,6 +137,64 @@ def create_directory(self,dirn):
                     print("Making %s" % dir_path)
                     bcf_utils.mkdir(dir_path)
 
+    def update_paths(self, base_path=None, new_path=None):
+        """
+        Update the paths stored in the analysis directory
+
+        Checks the paths stored in the analysis directory
+        metadata and parameter files, and updates them if
+        they're inconsistent with the current location.
+
+        By default the original 'base' path is taken from
+        the path stored in the analysis directory parameters,
+        and the new 'base' path is assumed to be the current
+        path for the analysis directory (these settings
+        are sensible if the directory has been relocated or
+        copied).
+
+        Arguments:
+            base_path (str): current 'base' directory path
+            (defaults to path stored in analysis directory
+            parameters)
+            new_path (str): new 'base' directory path
+            (defaults to the current path of the analysis
+            directory)
+        """
+        # Update paths in the top-level parameter file
+        # (if analysis dir has been moved or copied)
+        if base_path is None:
+            # Use stored path as original base path
+            base_path = self.params.analysis_dir
+        if new_path is None:
+            # Use current path as new base path
+            new_path = self.analysis_dir
+        if base_path != new_path:
+            print("Updating analysis directory paths in parameter file")
+            print(f"-- old base path: {base_path}")
+            print(f"-- new base path: {new_path}")
+            for p in ('analysis_dir',
+                      'primary_data_dir',
+                      'sample_sheet'):
+                if not self.params[p]:
+                    continue
+                self.params[p] = os.path.normpath(
+                    os.path.join(new_path,
+                                 os.path.relpath(self.params[p], base_path)))
+                print(f"...updated '{p}' (set to '{self.params[p]}'")
+            # Update paths in QC metadata in projects
+            for project in self.get_analysis_projects_from_dirs():
+                # Iterate through all project directories
+                for qc_dir in project.qc_dirs:
+                    qc_info = project.qc_info(qc_dir)
+                    qc_info['fastq_dir'] = os.path.normpath(
+                        os.path.join(new_path,
+                                     os.path.relpath(qc_info.fastq_dir,
+                                                     base_path)))
+                    print(f"...updated QC info for {project.name}/{qc_dir}")
+                    qc_info.save()
+            # Save the updated parameter data
+            self.save_parameters(force=True)
+
     def load_parameters(self,allow_save=True):
         """
         Load parameter values from file
@@ -477,6 +532,58 @@ def update_project_metadata_file(self,unaligned_dir=None,
         print("Updated project metadata file '%s'" %
               self.params.project_metadata)
 
+    def sync_project_metadata_file(self):
+        """
+        Synchronise 'projects.info' file with directory contents
+        """
+        # Load information from 'projects.info'
+        project_metadata = self.load_project_metadata()
+        # Comment out projects which don't exist on filesystem
+        save_required = False
+        for line in project_metadata:
+            # Iterate through the named projects
+            name = line['Project']
+            if name.startswith('#'):
+                # Commented out, ignore
+                continue
+            # Look for a matching project directory
+            project_dir = os.path.join(self.analysis_dir, name)
+            if not os.path.exists(project_dir):
+                print(f"Commenting out missing project '{name}'")
+                line['Project'] = f"#{name}"
+                save_required = True
+        if save_required:
+            project_metadata.save()
+        # Add any project directories without entries
+        save_required = False
+        projects = [line['Project'] for line in project_metadata]
+        for project in self.get_analysis_projects_from_dirs():
+            if project.name.endswith(".bak") \
+                    or project.name.endswith(".orig") \
+                    or project.name.endswith(".tmp"):
+                # Skip directories with extensions indicating they
+                # should be ignored
+                print(f"Not adding entry for unlisted project '{project.name}'")
+                continue
+            elif project.name == "undetermined":
+                # Skip undetermined
+                continue
+            elif project.name not in projects and f"#{project.name}" not in projects:
+                # Add new entry
+                print(f"Adding entry for unlisted project '{project.name}'")
+                project_metadata.add_project(project.name,
+                                             [s.name for s in project.samples],
+                                             user=project.info.user,
+                                             PI=project.info.PI,
+                                             organism=project.info.organism,
+                                             library_type=project.info.library_type,
+                                             sc_platform=project.info.single_cell_platform,
+                                             comments=project.info.comments)
+                save_required = True
+        # Save the updated project metadata if required
+        if save_required:
+            project_metadata.save()
+
     def detect_unaligned_dir(self):
         # Attempt to detect an existing 'bcl2fastq' or 'Unaligned' directory
         # containing data from bcl2fastq
@@ -920,6 +1027,65 @@ def get_analysis_projects_from_dirs(self,pattern=None,strict=False):
                 projects.append(test_project)
         return projects
 
+    def sync_project_metadata(self):
+        """
+        Update metadata stored in project dirs with 'projects.info'
+        """
+        # Load information from 'projects.info'
+        project_metadata = self.load_project_metadata()
+        save_required = False
+        for line in project_metadata:
+            # Iterate through the named projects
+            name = line['Project']
+            if name.startswith('#'):
+                # Commented out, ignore
+                continue
+            # Look for a matching project directory
+            project_dir = os.path.join(self.analysis_dir, name)
+            if os.path.exists(project_dir):
+                project = AnalysisProject(project_dir)
+                print(f"Checking metadata for project '{name}'")
+                # Synchronise metadata in projects with projects.info
+                metadata_items = dict(
+                    name=name,
+                    user=line['User'],
+                    PI=line['PI'],
+                    organism=line['Organism'],
+                    library_type=line['Library'],
+                    single_cell_platform=line['SC_Platform'],
+                    comments=line['Comments'],
+                    samples=project.sample_summary()
+                )
+                # Only update items where values differ
+                project_metadata_updated = False
+                for item in metadata_items:
+                    new_value = (metadata_items[item]
+                                 if metadata_items[item] != '.' else None)
+                    if project.info[item] != new_value:
+                        print("...updating '%s' => %r" % (item, new_value))
+                        project.info[item] = new_value
+                        project_metadata_updated = True
+                # Check paired-end info
+                project_info = AnalysisProjectInfo(project.info_file)
+                if project_info.paired_end != project.info.paired_end:
+                    print("...updating paired-end info")
+                    project_metadata_updated = True
+                # Save the updated project metadata if required
+                if project_metadata_updated:
+                    print(f"...saving project metadata for {project.name}")
+                    project.info.save()
+                # Update list of sample names in projects.info
+                sample_list = ','.join(sort_sample_names(
+                    [s.name for s in project.samples]))
+                if line['Samples'] != sample_list:
+                    print("...updating sample list in projects.info")
+                    line['Samples'] = sample_list
+                    save_required = True
+        # Save master project metadata
+        if save_required:
+            print("Saving projects.info")
+            project_metadata.save()
+
     def undetermined(self):
         # Return analysis project directory for undetermined indices
         # or None if not found

auto_process_ngs/commands/archive_cmd.py

@@ -10,11 +10,10 @@
 #######################################################################
 
 import os
-import time
 import logging
 from ..analysis import AnalysisDir
+from ..auto_processor import AutoProcess
 from ..metadata import AnalysisDirMetadata
-from ..metadata import AnalysisDirParameters
 from ..command import Command
 from ..commands.report_cmd import report_concise
 from .. import apps
@@ -412,37 +411,16 @@ def archive(ap,archive_dir=None,platform=None,year=None,
             sched.stop()
     # Perform final archiving operations
     if final:
-        # Update the final stored Fastq paths and metadata
-        # FIXME this is essentially duplicating functionality
-        # FIXME in the 'update' command
-        # FIXME (Also probably shouldn't update metadata for
-        # FIXME 'dry_run' mode?)
-        print("Updating stored paths and metadata")
+        # Paths to staging and final directories
         staged_analysis_dir = os.path.join(archive_dir,staging)
         archived_analysis_dir = os.path.join(archive_dir,final_dest)
-        parameter_file = os.path.join(staged_analysis_dir,
-                                      "auto_process.info")
-        if os.path.exists(parameter_file):
-            params = AnalysisDirParameters()
-            params.load(parameter_file,strict=False)
-            base_path = params.analysis_dir
-            print("Stored base path: %s" % base_path)
-            for p in ('analysis_dir',
-                      'primary_data_dir',
-                      'sample_sheet'):
-                if not params[p]:
-                    continue
-                params[p] = os.path.normpath(
-                    os.path.join(archived_analysis_dir,
-                                 os.path.relpath(params[p],
-                                                 base_path)))
-                print("...updated '%s' (set to '%s')" % (p,params[p]))
-            params.save()
+        # Update the final stored Fastq paths and metadata
+        if not dry_run:
+            AutoProcess(staged_analysis_dir).update_paths(
+                base_path=ap.params.analysis_dir,
+                new_path=archived_analysis_dir)
         else:
-            base_path = ap.analysis_dir
-            logger.warning("Unable to get old base path from parameters")
-            logger.warning("Using base path: %s (may be incorrect)" %
-                           base_path)
+            print("Updating paths skipped for dry run")
         # Run ID and reference
         metadata_file = os.path.join(staged_analysis_dir,
                                      "metadata.info")
@@ -456,7 +434,10 @@ def archive(ap,archive_dir=None,platform=None,year=None,
                 metadata['run_reference_id'] = ap.run_reference_id
                 print("...storing run reference ID ('%s')" %
                       metadata.run_reference_id)
-            metadata.save()
+            if not dry_run:
+                metadata.save()
+            else:
+                print("Run ID/reference ID not updated for dry run")
         # Project metadata and QC info
         analysis_dir =  AnalysisDir(staged_analysis_dir)
         # FIXME AnalysisDir.get_projects method might not get all
@@ -489,28 +470,16 @@ def archive(ap,archive_dir=None,platform=None,year=None,
                           project.name)
             # Save the updated information
             if project_info_updated:
-                project.info.save()
-            # QC metadata
-            # FIXME should do all QC dirs (not just the primary one)
-            qc_info = project.qc_info(project.qc_dir)
-            if qc_info.fastq_dir:
-                print("Project '%s': updating stored Fastq directory for QC" %
-                      project.name)
-                # FIXME could we just set it to the current Fastq path?
-                new_fastq_dir = os.path.normpath(
-                    os.path.join(archived_analysis_dir,
-                                 os.path.relpath(qc_info.fastq_dir,
-                                                 base_path)))
-                print("...updated Fastq directory: %s" % new_fastq_dir)
-                qc_info['fastq_dir'] = new_fastq_dir
                 if not dry_run:
-                    qc_info.save()
-                # Bail out if there was a problem
-                if retval != 0:
-                    if not force:
-                        raise Exception("Finalising archive failed")
-                    else:
-                        logger.warning("Finalising archive failed (ignored)")
+                    project.info.save()
+                else:
+                    print("Project metadata not updated for dry run")
+        # Bail out if there was a problem
+        if retval != 0:
+            if not force:
+                raise Exception("Finalising archive failed")
+            else:
+                logger.warning("Finalising archive failed (ignored)")
         # Complete archiving
         print("Moving to final location: %s" % final_dest)
         if not dry_run:

auto_process_ngs/commands/clone_cmd.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 #
 #     clone_cmd.py: implement auto process clone command
-#     Copyright (C) University of Manchester 2019 Peter Briggs
+#     Copyright (C) University of Manchester 2019-2026 Peter Briggs
 #
 #########################################################################
 
@@ -13,7 +13,7 @@
 import logging
 import shutil
 from ..analysis import AnalysisProject
-from ..metadata import AnalysisDirParameters
+from ..auto_processor import AutoProcess
 import bcftbx.utils as bcf_utils
 
 # Module specific logger
@@ -141,17 +141,5 @@ def clone(ap,clone_dir,copy_fastqs=False,exclude_projects=False):
     for subdir in ('logs','ScriptCode',):
         print("[Subdirectories] making %s" % subdir)
         bcf_utils.mkdir(os.path.join(clone_dir,subdir))
-    # Update the settings
-    parameter_file = os.path.join(clone_dir,
-                                  os.path.basename(ap.parameter_file))
-    params = AnalysisDirParameters(filen=os.path.join(
-        clone_dir,
-        os.path.basename(ap.parameter_file)))
-    for p in ("sample_sheet","primary_data_dir"):
-        if not params[p]:
-            continue
-        print("[Parameters] updating '%s'" % p)
-        params[p] = os.path.join(clone_dir,
-                                 os.path.relpath(params[p],
-                                                 ap.analysis_dir))
-    params.save()
+    # Update the paths
+    AutoProcess(clone_dir).update_paths()