Annotation Manifest For Each Analysis Omic Unit (#213)

* first pass to restructure the analyses' annotation manifest to be per omic unit annotated. Changes are stable and will continue to see if things continue to work when revising the testing

* changelog update

* first pass for large scale migration script

* Simplified logging messages for annotations;

Author: SeriousHorncat

CHANGELOG.md

@@ -10,6 +10,8 @@
 ### Features
 
 - The annotations page can select genes & variant of an analysis to view using the page URL's query parameters
+- Annotation manifests for analyses are now organized by omic unit, allowing more granular support for versioning of
+  annotations.
 
 ### Bugs
 

backend/src/core/annotation.py

@@ -23,7 +23,12 @@ def to_name_string(annotation_unit: AnnotationUnit):
     """
     Returns the annotation unit's genomic_unit and corresponding dataset.
     """
-    return f"{annotation_unit.get_genomic_unit().ljust(30)}{annotation_unit.get_dataset_name().ljust(60)}"
+    if not annotation_unit.version_calculated():
+        return f"{annotation_unit.get_genomic_unit().ljust(30)}{annotation_unit.get_dataset_name().ljust(105)}"
+
+    name_and_version = f"{annotation_unit.get_genomic_unit().ljust(30)}{annotation_unit.get_dataset_name().ljust(45)}"
+
+    return f"{name_and_version}{annotation_unit.get_dataset_source().ljust(30)}{str(annotation_unit.version).ljust(30)}"
 
 
 def annotation_log_label():
@@ -34,21 +39,24 @@ def annotation_log_label():
     return 'Annotation'.ljust(15)
 
 
-def format_annotation_logging(annotation_unit: AnnotationUnit, dataset=""):
+def format_annotation_logging(annotation_unit: AnnotationUnit, note=""):
     """
     Provides a formatted string for logging that is consistent with
     annotation unit's genomic_unit and corresponding dataset to the console
     The string is padded to make the logs uniform and easier to read.
     """
-    if dataset != "":
-        annotation_unit_string = f"{annotation_unit.get_genomic_unit().ljust(30)}{dataset.ljust(60)}"
-    else:
-        annotation_unit_string = to_name_string(annotation_unit)
+    annotation_unit_log_string = ""
 
     if "" != annotation_unit.analysis_name:
-        annotation_unit_string += f"'{annotation_unit.analysis_name}'".ljust(20)
+        annotation_unit_log_string += f"'{annotation_unit.analysis_name}'".ljust(20)
+
+    if annotation_unit is not None:
+        annotation_unit_log_string += f"{to_name_string(annotation_unit)}"
+
+    if note != "":
+        annotation_unit_log_string += f"{note.ljust(60)}"
 
-    return f"{annotation_log_label()}{annotation_unit_string}"
+    return f"{annotation_log_label()}{annotation_unit_log_string}"
 
 
 class AnnotationQueue:
@@ -104,7 +112,7 @@ def process_tasks(
         analysis_collection: AnalysisCollection
     ):
         """Processes items that have been added to the queue"""
-        logger.info("%s Processing annotation tasks queue ...", annotation_log_label())
+        logger.info("%s Processing annotation tasks queue...", annotation_log_label())
 
         processor = AnnotationProcess(annotation_queue, genomic_unit_collection, analysis_collection)
 
@@ -167,12 +175,31 @@ def process_annotation_unit(self, annotation_unit: AnnotationUnit):
         annotation task can be created. If the annotation unit is ready to annotate, it will be submitted to run
         on the task execeutor thread pool.
         """
-        if not annotation_unit.version_exists():
+
+        possible_manifest_entry = self.retrieve_manifest_entry_if_exist(annotation_unit)
+        if possible_manifest_entry is not None and not annotation_unit.version_calculated():
+            manifest_annotation_unit = self._create_temporary_annotation_unit(
+                annotation_unit.genomic_unit, possible_manifest_entry, annotation_unit.analysis_name,
+                annotation_unit.is_transcript_dataset()
+            )
+            if not self.genomic_unit_collection.annotation_exist(manifest_annotation_unit):
+                logger.error(
+                    '%s Manifest Annotation Does Not Exist...', format_annotation_logging(manifest_annotation_unit)
+                )
+                logger.error(
+                    '%s Remove Manifest Entry Manually...', format_annotation_logging(manifest_annotation_unit)
+                )
+            else:
+                logger.info('%s Manifest Annotation Exists...', format_annotation_logging(manifest_annotation_unit))
+                return
+
+        if not annotation_unit.version_calculated():
             self.handle_annotation_unit_version_calcuation(annotation_unit)
             return
 
         if self.genomic_unit_collection.annotation_exist(annotation_unit):
             logger.info('%s Annotation Exists...', format_annotation_logging(annotation_unit))
+            self.analysis_collection.add_dataset_to_manifest(annotation_unit.analysis_name, annotation_unit)
             return
 
         if annotation_unit.has_dependencies():
@@ -201,6 +228,7 @@ def on_task_complete(self, future):
         """
         task = self.annotation_task_futures[future]
         annotation_unit = task.annotation_unit
+        logger.info('%s Task Executed...', format_annotation_logging(annotation_unit))
 
         try:
             task_process_result = future.result()
@@ -216,14 +244,11 @@ def on_task_complete(self, future):
                 self.queue.put(annotation_unit)
             else:
                 for annotation in task.extract(task_process_result):
-                    logger.info(
-                        '%s Saving %s...',
-                        format_annotation_logging(annotation_unit,
-                                                  annotation_unit.get_dataset_name()), annotation['value']
-                    )
+                    logger.info('%s Saving %s...', format_annotation_logging(annotation_unit), annotation['value'])
 
                     self.genomic_unit_collection.annotate_genomic_unit(annotation_unit.genomic_unit, annotation)
                     self.analysis_collection.add_dataset_to_manifest(annotation_unit.analysis_name, annotation_unit)
+                logger.info('%s Complete...', format_annotation_logging(annotation_unit))
 
         except FileNotFoundError as error:
             logger.error('%s Exception [%s] Not Found [%s]', format_annotation_logging(annotation_unit), error, task)
@@ -275,6 +300,12 @@ def set_version_in_cache(self, version_cache_id: str, version):
         """Sets the version to be cached in the version cache"""
         self.version_cache[version_cache_id] = version
 
+    def retrieve_manifest_entry_if_exist(self, annotation_unit: AnnotationUnit):
+        """Queries the collection for an annotation unit's manifest entry. If none exists None is returned."""
+        return self.analysis_collection.get_manifest_dataset_config(
+            annotation_unit.analysis_name, annotation_unit.get_genomic_unit(), annotation_unit.get_dataset_name()
+        )
+
     def handle_annotation_unit_version_calcuation(self, annotation_unit):
         """
         Processes the annotation unit to derive the annotation unit's calculated version according to the configuration.
@@ -284,25 +315,24 @@ def handle_annotation_unit_version_calcuation(self, annotation_unit):
         version_cache_id = version_task.get_version_cache_id()
 
         if not self.is_version_cache_setup(version_cache_id):
-            logger.info('%s Creating Task To Version...', format_annotation_logging(annotation_unit))
+            logger.info('%s Creating Calculate Version Task...', format_annotation_logging(annotation_unit))
             self.setup_version_cache(version_cache_id)
             self.queue_task_in_tasks_worker(version_task)
             return
 
         if self.is_version_cached(version_cache_id):
             cached_version = self.version_cache[version_cache_id]
             annotation_unit.set_latest_version(cached_version)
-            logger.info(
-                '%s Version Gathered from Cache %s...', format_annotation_logging(annotation_unit), cached_version
-            )
+            logger.info('%s Version From Cache %s...', format_annotation_logging(annotation_unit), cached_version)
         self.queue.put(annotation_unit)
 
     def handle_annotation_unit_dependencies(self, annotation_unit: AnnotationUnit):
         """Retrieves the an annotation unit's dependencies if they exist."""
+
         missing_dependencies = annotation_unit.get_missing_dependencies()
         for missing_dataset_name in missing_dependencies:
             analysis_manifest_dataset = self.analysis_collection.get_manifest_dataset_config(
-                annotation_unit.analysis_name, missing_dataset_name
+                annotation_unit.analysis_name, annotation_unit.get_genomic_unit(), missing_dataset_name
             )
 
             if analysis_manifest_dataset is None:
@@ -321,7 +351,7 @@ def handle_annotation_unit_dependencies(self, annotation_unit: AnnotationUnit):
 
         if annotation_unit.if_transcript_needs_provisioning():
             transcript_id_manifest_dataset = self.analysis_collection.get_manifest_dataset_config(
-                annotation_unit.analysis_name, "transcript_id"
+                annotation_unit.analysis_name, annotation_unit.get_genomic_unit(), "transcript_id"
             )
 
             if transcript_id_manifest_dataset is not None:
@@ -337,9 +367,14 @@ def handle_annotation_unit_dependencies(self, annotation_unit: AnnotationUnit):
                 if transcript_id_dataset_saved:
                     annotation_unit.set_transcript_provisioned(True)
 
-    def _create_temporary_annotation_unit(self, genomic_unit, manifest_dataset, analysis_name):
+    def _create_temporary_annotation_unit(
+        self, genomic_unit, manifest_dataset, analysis_name: str, transcript_dataset: bool = False
+    ):
         """Private helper method to create a temporary annotation unit for finding within repository"""
         temporary = AnnotationUnit(genomic_unit, manifest_dataset, analysis_name)
         temporary.set_latest_version(manifest_dataset['version'])
 
+        if transcript_dataset:
+            temporary.dataset['transcript'] = True
+
         return temporary

backend/src/core/annotation_unit.py

@@ -124,14 +124,14 @@ def should_continue_annotation(self):
         delay_count = self.dataset['delay_count'] + 1 if 'delay_count' in self.dataset else 0
         self.dataset['delay_count'] = delay_count
 
-        is_delay_count_exceeding = self.dataset['delay_count'] > 10
+        is_delay_count_exceeding = self.dataset['delay_count'] > 20
 
         return not is_delay_count_exceeding
 
     def set_latest_version(self, version_details):
         """Sets the Annotation Unit with the version"""
         self.version = version_details
 
-    def version_exists(self):
+    def version_calculated(self):
         """Checks if the Annotation Unit is versioned or not"""
         return self.version != ""

backend/src/models/analysis.py

@@ -90,7 +90,7 @@ class Analysis(BaseAnalysis):
     sections: List[Section] = []
     discussions: List = []
     attachments: List = []
-    manifest: List = []
+    manifest: object = {}
 
     def units_to_annotate(self):
         """Returns the types of genomic units within the analysis"""

backend/src/repository/analysis_collection.py

@@ -134,31 +134,51 @@ def get_genomic_units(self, analysis_name: str):
 
     def add_dataset_to_manifest(self, analysis_name: str, annotation_unit: AnnotationUnit):
         """Adds this dataset and its version to this Analysis."""
-
         dataset = {
             annotation_unit.get_dataset_name(): {
                 'data_source': annotation_unit.get_dataset_source(), 'version': annotation_unit.version
             }
         }
 
-        updated_document = self.collection.find_one_and_update({"name": analysis_name},
-                                                               {"$addToSet": {"manifest": dataset}},
+        self.collection.find_one_and_update({
+            "name": analysis_name, "manifest.unit": {"$ne": annotation_unit.get_genomic_unit()}
+        }, {'$addToSet': {'manifest': {'unit': annotation_unit.get_genomic_unit(), 'manifest': []}}})
+
+        updated_document = self.collection.find_one_and_update({
+            "name": analysis_name, "manifest.unit": annotation_unit.get_genomic_unit()
+        }, {"$addToSet": {"manifest.$[entry].manifest": dataset}},
+                                                               array_filters=[{
+                                                                   'entry.unit': annotation_unit.get_genomic_unit()
+                                                               }],
                                                                return_document=ReturnDocument.AFTER)
 
         return updated_document['manifest']
 
-    def get_manifest_dataset_config(self, analysis_name: str, dataset_name: str):
+    def get_manifest_dataset_config(self, analysis_name: str, omic_unit: str, dataset_name: str):
         """ Returns an individual dataset manifest """
-        dataset_attribute = f"manifest.{dataset_name}"
 
-        projection = {"manifest.$": 1}
-        query = {"name": analysis_name, dataset_attribute: {'$exists': True}}
+        dataset_field = f"manifest.{dataset_name}"
+
+        query = {
+            "name": analysis_name, "manifest": {"$elemMatch": {"unit": omic_unit, dataset_field: {'$exists': True}}}
+        }
+
+        projection = {"manifest.manifest": 1, "manifest.unit.$": 1}
+
         analysis = self.collection.find_one(query, projection)
 
         if not analysis:
             return None
 
-        manifest_entry = next((dataset for dataset in analysis['manifest'] if dataset_name in dataset), None)
+        unit_manifest = next((
+            unit_manifest for unit_manifest in analysis['manifest']
+            if "unit" in unit_manifest and unit_manifest['unit'] == omic_unit
+        ), None)
+
+        if not unit_manifest:
+            return None
+
+        manifest_entry = next((dataset for dataset in unit_manifest['manifest'] if dataset_name in dataset), None)
 
         return {
             "data_set": dataset_name, "data_source": manifest_entry[dataset_name]['data_source'],

backend/src/routers/analysis_annotation_router.py

@@ -23,7 +23,7 @@ def get_annotations_by_gene(analysis_name, gene, repositories=Depends(database))
         raise HTTPException(status_code=404, detail=f"Gene'{gene}' annotations not found.")
 
     manifest = AnalysisDatasetManfiest(dataset_manifest)
-    annotations = manifest.retrieve_annotations(genomic_unit_json['annotations'])
+    annotations = manifest.retrieve_annotations(gene, genomic_unit_json['annotations'])
 
     return annotations
 
@@ -45,11 +45,11 @@ def get_annotations_by_hgvs_variant(analysis_name: str, variant: str, repositori
         raise HTTPException(status_code=404, detail=f"Variant'{variant}' annotations not found.")
 
     manifest = AnalysisDatasetManfiest(dataset_manifest)
-    annotations = manifest.retrieve_annotations(genomic_unit_json['annotations'])
+    annotations = manifest.retrieve_annotations(variant, genomic_unit_json['annotations'])
 
     transcript_annotation_list = []
     for transcript_annotation in genomic_unit_json['transcripts']:
-        transcript_annotations = manifest.retrieve_annotations(transcript_annotation['annotations'])
+        transcript_annotations = manifest.retrieve_annotations(variant, transcript_annotation['annotations'])
         transcript_annotation_list.append(transcript_annotations)
 
     return {**annotations, "transcripts": transcript_annotation_list}
@@ -75,7 +75,7 @@ def __init__(self, analysis_dataset_manifest):
         """
         self.manifest = analysis_dataset_manifest
 
-    def retrieve_annotations(self, unit_annotations):
+    def retrieve_annotations(self, omic_unit: str, unit_annotations):
         """
         Extracts annotations from the provided list of unit annotations and returns a dictionary
         of datasets and their corresponding values.
@@ -94,20 +94,27 @@ def retrieve_annotations(self, unit_annotations):
         for annotation_json in unit_annotations:
             for dataset in annotation_json:
                 if len(annotation_json[dataset]) > 0:
-                    analysis_dataset = self.get_value_for_dataset(dataset, annotation_json[dataset])
+                    analysis_dataset = self.get_value_for_dataset(dataset, omic_unit, annotation_json[dataset])
                     annotations[dataset] = analysis_dataset[
                         'value'] if analysis_dataset is not None else annotation_json[dataset][0]['value']
         return annotations
 
-    def get_value_for_dataset(self, dataset_name: str, annotation_json_list: list):
+    def get_value_for_dataset(self, dataset_name: str, omic_unit: str, annotation_json_list: list):
         """
         Retrieves the annotation according to the analysis' dataset manifest entry matching the dataset's name,
         'data_source', and 'version'. None is returned when there isn't an entry in the manifest.
         """
-        dataset_config = next((configuration for configuration in self.manifest if dataset_name in configuration), None)
+
+        omic_manifest = next((manifest for manifest in self.manifest if manifest['unit'] == omic_unit), None)
+        if omic_manifest is None:
+            return None
+
+        dataset_config = next(
+            (configuration for configuration in omic_manifest['manifest'] if dataset_name in configuration), None
+        )
 
         if dataset_config is None:
-            return dataset_config
+            return None
 
         configuration = dataset_config[dataset_name]