use predetermined filters/imputed_sex paths

Author: bpblanken

v03_pipeline/lib/model/environment.py

@@ -2,10 +2,6 @@
 from dataclasses import dataclass
 
 # NB: using os.environ.get inside the dataclass defaults gives a lint error.
-ACCESS_PRIVATE_REFERENCE_DATASETS = (
-    os.environ.get('ACCESS_PRIVATE_REFERENCE_DATASETS') == '1'
-)
-REFERENCE_DATA_AUTO_UPDATE = os.environ.get('REFERENCE_DATA_AUTO_UPDATE') == '1'
 HAIL_TMPDIR = os.environ.get('HAIL_TMPDIR', '/tmp')  # noqa: S108
 HAIL_SEARCH_DATA = os.environ.get('HAIL_SEARCH_DATA', '/hail-search-data')
 LOADING_DATASETS = os.environ.get('LOADING_DATASETS', '/seqr-loading-temp')
@@ -19,22 +15,32 @@
 )
 VEP_CONFIG_PATH = os.environ.get('VEP_CONFIG_PATH', None)
 VEP_CONFIG_URI = os.environ.get('VEP_CONFIG_URI', None)
-SHOULD_REGISTER_ALLELES = os.environ.get('SHOULD_REGISTER_ALLELES') == '1'
+
+# Allele registry secrets :/
 ALLELE_REGISTRY_SECRET_NAME = os.environ.get('ALLELE_REGISTRY_SECRET_NAME', None)
 PROJECT_ID = os.environ.get('PROJECT_ID', None)
 
+# Feature Flags
+ACCESS_PRIVATE_REFERENCE_DATASETS = (
+    os.environ.get('ACCESS_PRIVATE_REFERENCE_DATASETS') == '1'
+)
+EXPECT_WES_FILTERS = os.environ.get('EXPECT_WES_FILTERS') == '1'
+REFERENCE_DATA_AUTO_UPDATE = os.environ.get('REFERENCE_DATA_AUTO_UPDATE') == '1'
+SHOULD_REGISTER_ALLELES = os.environ.get('SHOULD_REGISTER_ALLELES') == '1'
+
 
 @dataclass
 class Env:
     ACCESS_PRIVATE_REFERENCE_DATASETS: bool = ACCESS_PRIVATE_REFERENCE_DATASETS
     ALLELE_REGISTRY_SECRET_NAME: str | None = ALLELE_REGISTRY_SECRET_NAME
-    REFERENCE_DATA_AUTO_UPDATE: bool = REFERENCE_DATA_AUTO_UPDATE
+    EXPECT_WES_FILTERS: bool = EXPECT_WES_FILTERS
     HAIL_TMPDIR: str = HAIL_TMPDIR
     HAIL_SEARCH_DATA: str = HAIL_SEARCH_DATA
     LOADING_DATASETS: str = LOADING_DATASETS
     PRIVATE_REFERENCE_DATASETS: str = PRIVATE_REFERENCE_DATASETS
     PROJECT_ID: str | None = PROJECT_ID
     REFERENCE_DATASETS: str = REFERENCE_DATASETS
+    REFERENCE_DATA_AUTO_UPDATE: bool = REFERENCE_DATA_AUTO_UPDATE
     SHOULD_REGISTER_ALLELES: bool = SHOULD_REGISTER_ALLELES
     VEP_CONFIG_PATH: str | None = VEP_CONFIG_PATH
     VEP_CONFIG_URI: str | None = VEP_CONFIG_URI

v03_pipeline/lib/paths.py

@@ -1,5 +1,6 @@
 import hashlib
 import os
+import re
 
 from v03_pipeline.lib.model import (
     AccessControl,
@@ -9,6 +10,7 @@
     PipelineVersion,
     ReferenceDatasetCollection,
     ReferenceGenome,
+    SampleType,
 )
 
 
@@ -73,6 +75,22 @@ def family_table_path(
     )
 
 
+def imputed_sex_path(
+    reference_genome: ReferenceGenome,
+    dataset_type: DatasetType,
+    callset_path: str,
+) -> str:
+    return os.path.join(
+        _v03_pipeline_prefix(
+            Env.LOADING_DATASETS,
+            reference_genome,
+            dataset_type,
+        ),
+        'imputed_sex',
+        f'{hashlib.sha256(callset_path.encode("utf8")).hexdigest()}.tsv',
+    )
+
+
 def imported_callset_path(
     reference_genome: ReferenceGenome,
     dataset_type: DatasetType,
@@ -198,6 +216,25 @@ def sex_check_table_path(
     )
 
 
+def valid_filters_path(
+    dataset_type: DatasetType,
+    sample_type: SampleType,
+    callset_path: str,
+) -> str | None:
+    if (
+        not Env.EXPECT_WES_FILTERS
+        or dataset_type != DatasetType.SNV_INDEL
+        or sample_type != SampleType.WES
+        or 'part_one_outputs' not in callset_path
+    ):
+        return None
+    return re.sub(
+        'part_one_outputs/.*$',
+        'part_two_outputs/*.filtered.*.vcf.gz',
+        callset_path,
+    )
+
+
 def valid_reference_dataset_collection_path(
     reference_genome: ReferenceGenome,
     dataset_type: DatasetType,

v03_pipeline/lib/paths_test.py

@@ -6,18 +6,21 @@
     DatasetType,
     ReferenceDatasetCollection,
     ReferenceGenome,
+    SampleType,
 )
 from v03_pipeline.lib.paths import (
     cached_reference_dataset_query_path,
     family_table_path,
     imported_callset_path,
+    imputed_sex_path,
     lookup_table_path,
     metadata_for_run_path,
     new_variants_table_path,
     project_table_path,
     relatedness_check_table_path,
     remapped_and_subsetted_callset_path,
     sex_check_table_path,
+    valid_filters_path,
     valid_reference_dataset_collection_path,
     variant_annotations_table_path,
 )
@@ -54,6 +57,24 @@ def test_family_table_path(self) -> None:
                 'gs://seqr-datasets/v03/GRCh37/SNV_INDEL/families/franklin.ht',
             )
 
+    def test_valid_filters_path(self) -> None:
+        self.assertEqual(
+            valid_filters_path(
+                SampleType.WES,
+                'gs://bucket/RDG_Broad_WES_Internal_Oct2023/part_one_outputs/chr*/*.vcf.gz',
+            ),
+            None,
+        )
+        with patch('v03_pipeline.lib.paths.Env') as mock_env:
+            mock_env.EXPECT_WES_FILTERS = True
+            self.assertEqual(
+                valid_filters_path(
+                    SampleType.WES,
+                    'gs://bucket/RDG_Broad_WES_Internal_Oct2023/part_one_outputs/chr*/*.vcf.gz',
+                ),
+                'gs://bucket/RDG_Broad_WES_Internal_Oct2023/part_two_outputs/*.filtered.*.vcf.gz',
+            )
+
     def test_project_table_path(self) -> None:
         self.assertEqual(
             project_table_path(
@@ -162,6 +183,16 @@ def test_imported_callset_path(self) -> None:
             '/seqr-loading-temp/v03/GRCh38/SNV_INDEL/imported_callsets/ead56bb177a5de24178e1e622ce1d8beb3f8892bdae1c925d22ca0af4013d6dd.mt',
         )
 
+    def test_imputed_sex_path(self) -> None:
+        self.assertEqual(
+            imputed_sex_path(
+                ReferenceGenome.GRCh38,
+                DatasetType.SNV_INDEL,
+                'gs://abc.efg/callset.vcf.gz',
+            ),
+            '/seqr-loading-temp/v03/GRCh38/SNV_INDEL/imputed_sex/ead56bb177a5de24178e1e622ce1d8beb3f8892bdae1c925d22ca0af4013d6dd.tsv',
+        )
+
     def test_new_variants_table_path(self) -> None:
         self.assertEqual(
             new_variants_table_path(

v03_pipeline/lib/tasks/base/base_loading_run_params.py

@@ -10,16 +10,6 @@ class BaseLoadingRunParams(luigi.Task):
     # but nothing else.
     sample_type = luigi.EnumParameter(enum=SampleType)
     callset_path = luigi.Parameter()
-    # HINT: OptionalParameter vs Parameter is significant here.
-    # The default Parameter will case `None` to the string "None".
-    imputed_sex_path = luigi.OptionalParameter(
-        default=None,
-        description='Optional path to a tsv of imputed sex values from the DRAGEN GVS pipeline.',
-    )
-    filters_path = luigi.OptionalParameter(
-        default=None,
-        description='Optional path to part two outputs from callset (VCF shards containing filter information)',
-    )
     ignore_missing_samples_when_remapping = luigi.BoolParameter(
         default=False,
         parsing=luigi.BoolParameter.EXPLICIT_PARSING,

v03_pipeline/lib/tasks/write_imported_callset.py

@@ -16,12 +16,13 @@
     validate_sample_type,
 )
 from v03_pipeline.lib.misc.vets import annotate_vets
-from v03_pipeline.lib.model import CachedReferenceDatasetQuery
+from v03_pipeline.lib.model import CachedReferenceDatasetQuery, DatasetType, SampleType
 from v03_pipeline.lib.model.environment import Env
 from v03_pipeline.lib.paths import (
     cached_reference_dataset_query_path,
     imported_callset_path,
     sex_check_table_path,
+    valid_filters_path,
 )
 from v03_pipeline.lib.tasks.base.base_loading_run_params import BaseLoadingRunParams
 from v03_pipeline.lib.tasks.base.base_write import BaseWriteTask
@@ -53,10 +54,18 @@ def output(self) -> luigi.Target:
 
     def requires(self) -> list[luigi.Task]:
         requirements = []
-        if self.filters_path:
+        if (
+            Env.EXPECT_WES_FILTERS
+            and self.dataset_type == DatasetType.SNV_INDEL
+            and self.sample_type == SampleType.WES
+        ):
             requirements = [
                 *requirements,
-                CallsetTask(self.filters_path),
+                CallsetTask(
+                    valid_filters_path(
+                        self.dataset_type, self.sample_type, self.callset_path,
+                    ),
+                ),
             ]
         if self.validate and self.dataset_type.can_run_validation:
             requirements = [
@@ -108,11 +117,14 @@ def additional_row_fields(self, mt):
         }
 
     def create_table(self) -> hl.MatrixTable:
+        filters_path = valid_filters_path(
+            self.dataset_type, self.sample_type, self.callset_path,
+        )
         mt = import_callset(
             self.callset_path,
             self.reference_genome,
             self.dataset_type,
-            self.filters_path,
+            filters_path,
         )
         mt = select_relevant_fields(
             mt,
@@ -174,6 +186,6 @@ def create_table(self) -> hl.MatrixTable:
             )
         return mt.annotate_globals(
             callset_path=self.callset_path,
-            filters_path=self.filters_path or hl.missing(hl.tstr),
+            filters_path=filters_path or hl.missing(hl.tstr),
             sample_type=self.sample_type.value,
         )

v03_pipeline/lib/tasks/write_sex_check_table.py

@@ -2,14 +2,13 @@
 import luigi
 
 from v03_pipeline.lib.misc.io import import_imputed_sex
-from v03_pipeline.lib.paths import sex_check_table_path
+from v03_pipeline.lib.paths import imputed_sex_path, sex_check_table_path
 from v03_pipeline.lib.tasks.base.base_write import BaseWriteTask
 from v03_pipeline.lib.tasks.files import GCSorLocalTarget, RawFileTask
 
 
 class WriteSexCheckTableTask(BaseWriteTask):
     callset_path = luigi.Parameter()
-    imputed_sex_path = luigi.Parameter()
 
     def output(self) -> luigi.Target:
         return GCSorLocalTarget(
@@ -21,7 +20,13 @@ def output(self) -> luigi.Target:
         )
 
     def requires(self) -> luigi.Task:
-        return RawFileTask(self.imputed_sex_path)
+        return RawFileTask(
+            imputed_sex_path(
+                self.reference_genome,
+                self.dataset_type,
+                self.callset_path,
+            ),
+        )
 
     def create_table(self) -> hl.Table:
         return import_imputed_sex(self.input().path)