CPG-only: restructure output paths

Dockerfile

@@ -61,7 +61,7 @@ COPY --from=bcftools_compiler /bcftools_install/usr/local/lib/ /usr/local/lib/
 RUN ldconfig
 
 ARG ECHTVAR_VERSION=v0.2.2
-ARG VERSION=11.0.1
+ARG VERSION=11.0.2
 
 RUN wget -q -O /bin/echtvar "https://github.com/brentp/echtvar/releases/download/${ECHTVAR_VERSION}/echtvar" && \
     chmod +x /bin/echtvar

README.md

@@ -69,7 +69,7 @@ There are two primary workflows:
 To build the Docker image:
 
 ```
-docker build -t talos:11.0.1 .
+docker build -t talos:11.0.2 .
 ```
 
 ### **2. Download Annotation Resources**

docs/getting-started.md

@@ -25,7 +25,7 @@ You will need:
 Build the Talos Docker image locally:
 
 ```bash
-docker build -t talos:11.0.1 .
+docker build -t talos:11.0.2 .
 ```
 
 ---

nextflow.config

@@ -16,8 +16,8 @@ params {
     large_files = "large_files"
     ref_genome = "${params.large_files}/ref.fa"
 
-    // Docker container - "docker build -t talos:11.0.1 ."
-    container = 'talos:11.0.1'
+    // Docker container - "docker build -t talos:11.0.2 ."
+    container = 'talos:11.0.2'
 
     // MANE transcript resource
     mane = "${params.large_files}/MANE.GRCh38.v1.5.summary.txt.gz"

nextflow_schema.json

@@ -33,7 +33,7 @@
         },
         "container": {
           "type": "string",
-          "default": "talos:11.0.1",
+          "default": "talos:11.0.2",
           "description": "Docker container used by processes."
         },
         "mane": {

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "hatchling.build"
 name='talos'
 description='Centre for Population Genomics Variant Prioritisation'
 readme = "README.md"
-version="11.0.1"
+version="11.0.2"
 requires-python = ">=3.10,<3.12"
 license-files = ["LICENSE"]
 classifiers=[
@@ -124,9 +124,10 @@ hail = ["hail", "hailtop"]
 ## - we don't need generic cpg-flow arguments for every Stage, but need to fit the required method signature
 "src/talos/cpg_internal_scripts/annotation_stages.py" = ["ARG002"]
 "src/talos/cpg_internal_scripts/talos_stages.py" = ["ARG002"]
+"test/test_cpg_flow_utils.py" = ['ANN202', 'ARG001']
 
 [tool.bumpversion]
-current_version = "11.0.1"
+current_version = "11.0.2"
 parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)"
 serialize = ["{major}.{minor}.{patch}"]
 commit = true

src/talos/cpg_internal_scripts/CPG_Dockerfile

@@ -16,7 +16,7 @@ FROM base AS bcftools_compiler
 
 # BCFtools 1.22+ doesn't do any inference on chr-prefixes during CSQ, so the mismatch between fasta/vcf, and the gff3 (no chr-prefix) needs to be handled explicitly
 # BCFtools >=1.22 is required internally as it's the first version with a built-in Mitochondrial lookup table.
-# AS OF 11.0.1, Talos is building BCFtools from a private fork. This fork contains a single change - csq applies annotations
+# AS OF 11.0.2, Talos is building BCFtools from a private fork. This fork contains a single change - csq applies annotations
 # to both coding and non-coding genes in the event of overlapping genes. By default BCFtools skips non-coding gene annotation
 # if a coding transcript consequence was detected, but in practice this is masking clinically relevant non-coding gene variation
 # in cases where the non-coding gene overlaps with a non-clinically relevant coding gene.
@@ -53,7 +53,7 @@ RUN apt-get update && apt-get install --no-install-recommends -y \
 FROM base AS talos
 
 ARG ECHTVAR_VERSION=v0.2.2
-ARG VERSION=11.0.1
+ARG VERSION=11.0.2
 
 COPY --from=bcftools_compiler /bcftools_install/usr/local/bin/* /usr/local/bin/
 COPY --from=bcftools_compiler /bcftools_install/usr/local/libexec/bcftools/* /usr/local/libexec/bcftools/

src/talos/cpg_internal_scripts/annotation_stages.py

@@ -65,9 +65,8 @@ def does_final_file_path_exist(cohort: targets.Cohort) -> bool:
 
     return utils.exists(
         cpg_flow_utils.generate_dataset_prefix(
-            dataset=cohort.dataset.name,
+            cohort=cohort,
             stage_name='AnnotateSpliceAi',
-            hash_value=cohort.id,
         )
         / f'{cohort.id}.mt',
     )
@@ -111,10 +110,9 @@ class ExtractVcfFromMt(stage.CohortStage):
 
     def expected_outputs(self, cohort: targets.Cohort) -> dict[str, Path | str]:
         temp_prefix = cpg_flow_utils.generate_dataset_prefix(
-            dataset=cohort.dataset.name,
+            cohort=cohort,
             category='tmp',
             stage_name=self.name,
-            hash_value=cohort.id,
         )
         return {
             'id_file': str(temp_prefix / f'{cohort.id}-SG-ids.txt'),
@@ -145,10 +143,9 @@ class AnnotateUsingEchtvar(stage.CohortStage):
 
     def expected_outputs(self, cohort: targets.Cohort) -> dict[str, Path | str]:
         temp_prefix = cpg_flow_utils.generate_dataset_prefix(
-            dataset=cohort.dataset.name,
+            cohort=cohort,
             category='tmp',
             stage_name=self.name,
-            hash_value=cohort.id,
         )
         return {
             'success': temp_prefix / 'success.txt',
@@ -186,10 +183,9 @@ class AnnotateWithBcftoolsCsq(stage.CohortStage):
 
     def expected_outputs(self, cohort: targets.Cohort) -> dict[str, Path | str]:
         temp_prefix = cpg_flow_utils.generate_dataset_prefix(
-            dataset=cohort.dataset.name,
+            cohort=cohort,
             category='tmp',
             stage_name=self.name,
-            hash_value=cohort.id,
         )
         return {
             'success': temp_prefix / 'success.txt',
@@ -223,10 +219,9 @@ def queue_jobs(self, cohort: targets.Cohort, inputs: stage.StageInput) -> stage.
 class AnnotatedVcfIntoMt(stage.CohortStage):
     def expected_outputs(self, cohort: targets.Cohort) -> dict[str, Path | str]:
         temp_prefix = cpg_flow_utils.generate_dataset_prefix(
-            dataset=cohort.dataset.name,
+            cohort=cohort,
             category='tmp',
             stage_name=self.name,
-            hash_value=cohort.id,
         )
         return {
             'success': temp_prefix / 'success.txt',
@@ -250,10 +245,9 @@ def queue_jobs(self, cohort, inputs: stage.StageInput) -> stage.StageOutput:
             bcftools_template=bcftools_template,
             checkpoint=str(
                 cpg_flow_utils.generate_dataset_prefix(
-                    dataset=cohort.dataset.name,
+                    cohort=cohort,
                     category='tmp',
                     stage_name=self.name,
-                    hash_value=cohort.id,
                 )
                 / f'{cohort.id}_{{part}}annotation_checkpoint',
             ),
@@ -274,9 +268,8 @@ class AnnotateSpliceAi(stage.CohortStage):
     def expected_outputs(self, cohort: targets.Cohort) -> Path:
         return (
             cpg_flow_utils.generate_dataset_prefix(
-                dataset=cohort.dataset.name,
+                cohort=cohort,
                 stage_name=self.name,
-                hash_value=cohort.id,
             )
             / f'{cohort.id}.mt'
         )

src/talos/cpg_internal_scripts/build_report_index.py

@@ -16,6 +16,7 @@
 
 from metamist.graphql import gql, query
 
+COHORT_RE = re.compile(r'COH\d+')
 DATE_REGEX = re.compile(r'(\d{4}-\d{2}-\d{2})')
 
 JINJA_TEMPLATE_DIR = Path(__file__).absolute().parent.parent / 'templates'
@@ -56,9 +57,8 @@ class Report:
     dataset: str
     address: str
     is_exome: bool
-    is_long_read: bool
+    is_long_read: bool | str
     date: str
-    title: str
 
 
 @lru_cache(1)
@@ -72,21 +72,22 @@ def get_my_projects() -> set[str]:
     return all_projects
 
 
-def get_project_analyses(project: str) -> dict[tuple[bool, bool], str]:
+def get_project_analyses(project: str) -> dict[tuple[bool | str, bool], str]:
     """
     Find all the active analysis entries for this project, subdivide the analyses by long/short read and exome/genome
 
-    Create a dictionary indexed on a double boolean key:
-    - the file is long_read
+    Create a dictionary indexed on a double key:
+    - the file is long_read (either False, or the Cohort ID)
     - the file is exome
 
     This was chosen as in the CPG infrastructure, 'genome' is the default, so 'exome' is in the path if relevant.
-    Likewise for short-read vs. "long-read". This is not a perfect solution, but it works for now.
+    Long-read is a little more tricky, as we maintain separate callsets for each sequencer/library prep, so we can have
+    several live long-read analyses for a single project
 
     Also... you can use a tuple as a dictionary key and that's cool.
     """
 
-    project_reports: dict[tuple[bool, bool], str] = {}
+    project_reports: dict[tuple[bool | str, bool], str] = {}
 
     all_analyses = query(REPORT_QUERY, variables={'project': project})['project']['analyses']
     for analysis in all_analyses:
@@ -98,44 +99,50 @@ def get_project_analyses(project: str) -> dict[tuple[bool, bool], str]:
 
         output_path = outputs['path']
 
-        long = 'long_read' in output_path
+        if 'long_read' in output_path:
+            # bridging statement until all reports are reissued with new paths
+            try:
+                long = COHORT_RE.findall(output_path)[0]
+            except IndexError:
+                long = True
+        else:
+            long = False
+
         exome = 'exome' in output_path
 
         project_reports[(long, exome)] = output_path
 
     return project_reports
 
 
-def main(dataset: str = 'aip') -> None:
+def main(dataset: str) -> None:
     """
     Finds all existing reports, generates an HTML file as an index page.
     Args:
         dataset (str): The dataset to generate the index for, defaults to 'aip' for legacy reasons.
     """
 
-    parsed_reports = {cohort: get_project_analyses(cohort) for cohort in get_my_projects()}
+    parsed_reports = {each_dataset: get_project_analyses(each_dataset) for each_dataset in get_my_projects()}
 
     report_list: list[Report] = []
 
-    for cohort, cohort_results in parsed_reports.items():
-        for (long_read, exome), report_path in cohort_results.items():
+    for each_dataset, dataset_results in parsed_reports.items():
+        for (long_read, exome), report_path in dataset_results.items():
             # general - only one of these
             if report_path:
                 this_file_name = Path(report_path).name
                 trimmed_path = report_path.rstrip(this_file_name).rstrip('/')
 
                 for entry in list(map(str, to_anypath(trimmed_path).glob('*.html'))):
-                    report_address = entry.replace(WEB_BASE.format(cohort), WEB_URL_BASE.format(cohort))
-                    report_name = entry.split('/')[-1]
+                    report_address = entry.replace(WEB_BASE.format(each_dataset), WEB_URL_BASE.format(each_dataset))
                     if report_date := DATE_REGEX.search(report_address):
                         report_list.append(
                             Report(
-                                dataset=cohort,
+                                dataset=each_dataset,
                                 address=report_address,
                                 is_exome=exome,
                                 is_long_read=long_read,
                                 date=report_date.group(1),
-                                title=report_name,
                             ),
                         )
 
@@ -165,8 +172,8 @@ def cli_main():
     """
     Command line interface for the script.
     """
-    parser = ArgumentParser(description='Generate an index page for AIP reports')
-    parser.add_argument('--dataset', help='Dataset for the index page', default='aip')
+    parser = ArgumentParser(description='Generate an index page for Talos reports')
+    parser.add_argument('--dataset', help='Dataset for the index page')
     args = parser.parse_args()
     main(dataset=args.dataset)
 

src/talos/cpg_internal_scripts/cpg_flow_utils.py

@@ -2,7 +2,7 @@
 
 import loguru
 
-from cpg_flow import workflow
+from cpg_flow import targets, workflow
 from cpg_utils import Path, config, to_path
 from metamist import graphql
 
@@ -72,23 +72,24 @@ def check_for_dataset_centric_cohorts() -> bool:
 
 @cache
 def generate_dataset_prefix(
-    dataset: str,
+    cohort: targets.Cohort | None = None,
+    dataset: str | None = None,
     category: str | None = None,
     stage_name: str | None = None,
     hash_value: str | None = None,
 ) -> Path:
     """
-    Generate a dictionary of prefixes for the current workflow and Stage.
+    Generate a prefix for the current workflow and Stage.
     Needed because CPG-Flow currently lacks the granularity we need for both exome/genome and short/long read
     This is intended to generate the exact same prefix as CPG-Flow would generate, so that we continue previous work
     """
 
-    # mandatory value in a cpg-flow config
-    workflow_name = config.config_retrieve(['workflow', 'name'], 'talos')
+    if cohort is None and dataset is None:
+        raise RuntimeError('Must populate either cohort or dataset when calling generate_dataset_prefix')
 
-    # generated from the included samples in the workflow
-    # or passed directly if provided, e.g. to target outputs to a date-specific folder, not a callset-specific one
-    hash_element = hash_value or workflow.get_workflow().output_version
+    # allow for a missing Cohort (index page case)
+    cohort_id = cohort.id if cohort else None
+    dataset = cohort.dataset.name if cohort else dataset
 
     # allow subdivision by short/long read, and exome/genome
     # the current protocol here is to treat short read and genome as standard, and insert clarifying elements if needed
@@ -97,7 +98,11 @@ def generate_dataset_prefix(
 
     # line up all the elements into an ordered list, and then join the non-None elements
     suffix = '/'.join(
-        [x for x in [long_read_element, exome_element, workflow_name, hash_element, stage_name] if isinstance(x, str)],
+        [
+            x
+            for x in [long_read_element, exome_element, 'talos', cohort_id, hash_value, stage_name]
+            if isinstance(x, str)
+        ],
     )
 
     return to_path(config.dataset_path(suffix=suffix, dataset=dataset, category=category))