scripts/parse_tobwgs_csv.py
def find_gvcf(self, sample_id: str, cpg_id: Optional[str] = None) -> Optional[str]:
        """
        Find GVCF for the sample.
        """
        extension = 'g.vcf.gz'

        search_locations = [f'gs://cpg-tob-wgs-main-upload/{sample_id}.{extension}']

        if cpg_id:
            # Sample was added before and CPG ID is known, so can search the locations
            # where downstream upload processor and pipelines might have moved files.
            search_locations += [
                # After GVCFs were processed with joint-calling, staging ones
                # are moved to the archive:
                f'gs://cpg-tob-wgs-main-archive/{SOURCE}/gvcf/staging/{cpg_id}.{extension}',
                # Upload processor moves GVCFs from upload to the main bucket
                # as "staging" for processing by joint-calling:
                f'gs://cpg-tob-wgs-main/gvcf/staging/{cpg_id}.{extension}',
            ]

        for fpath in search_locations:
            if self.file_exists(fpath):
                return fpath

        return None

        Find GVCF for the sample.
        
Find GVCF for the sample.
Find GVCF for the sample .
def find_gvcf(self, sample_id: str, cpg_id: Optional[str] = None) -> Optional[str]:
        extension = 'g.vcf.gz'
        search_locations = [f'gs://cpg-tob-wgs-main-upload/{sample_id}.{extension}']
        if cpg_id:
            search_locations += [
                f'gs://cpg-tob-wgs-main-archive/{SOURCE}/gvcf/staging/{cpg_id}.{extension}',
                f'gs://cpg-tob-wgs-main/gvcf/staging/{cpg_id}.{extension}',
            ]
        for fpath in search_locations:
            if self.file_exists(fpath):
                return fpath
        return None
def find_gvcf ( self , sample_id : str , cpg_id : Optional [ str ] = None ) -> Optional [ str ] : extension = 'g.vcf.gz' search_locations = [ f'gs://cpg-tob-wgs-main-upload/{sample_id}.{extension}' ] if cpg_id : search_locations += [ f'gs://cpg-tob-wgs-main-archive/{SOURCE}/gvcf/staging/{cpg_id}.{extension}' , f'gs://cpg-tob-wgs-main/gvcf/staging/{cpg_id}.{extension}' , ] for fpath in search_locations : if self . file_exists ( fpath ) : return fpath return None
Find GVCF for the sample.
Find GVCF for the sample .
"""
        Find GVCF for the sample.
        """ # Sample was added before and CPG ID is known, so can search the locations # where downstream upload processor and pipelines might have moved files. # After GVCFs were processed with joint-calling, staging ones # are moved to the archive: # Upload processor moves GVCFs from upload to the main bucket # as "staging" for processing by joint-calling:
self
sample_id
str
cpg_id
Optional[str]