test/test_joint_calling_workflow.py
def _submit_analyses(samples: List, output_project: str, a_type: str):
    """
    Add or update analyses. Iterate over completed analyses,
    and submit next-step analyses
    """

    if a_type in ['gvcf', 'cram']:
        for s in samples:
            if a_type == 'gvcf':
                cram_analysis = aapi.get_latest_analysis_for_samples_and_type(
                    project=output_project,
                    analysis_type=AnalysisType('cram'),
                    request_body=[s['id']],
                )
                assert cram_analysis, s

            # completed_analysis = latest_by_type_and_sids.get(('cram', (s['id'],))),
            am = AnalysisModel(
                type=AnalysisType(a_type),
                output=f'result.{a_type}',
                status=AnalysisStatus('queued'),
                sample_ids=[s['id']],
            )
            aapi.create_new_analysis(project=output_project, analysis_model=am)

    elif a_type == 'joint-calling':
        for s in samples:
            gvcf_analysis = aapi.get_latest_analysis_for_samples_and_type(
                project=output_project,
                analysis_type=AnalysisType('gvcf'),
                request_body=[s['id']],
            )
            assert gvcf_analysis, s

        am = AnalysisModel(
            sample_ids=[s['id'] for s in samples],
            type=AnalysisType('joint-calling'),
            output='joint-called.vcf',
            status=AnalysisStatus('queued'),
        )
        aapi.create_new_analysis(project=output_project, analysis_model=am)

    Add or update analyses. Iterate over completed analyses,
    and submit next-step analyses
    
Add or update analyses. Iterate over completed analyses,
and submit next-step analyses
Add or update analyses . Iterate over completed analyses and submit next - step analyses
def _submit_analyses(samples: List, output_project: str, a_type: str):
    if a_type in ['gvcf', 'cram']:
        for s in samples:
            if a_type == 'gvcf':
                cram_analysis = aapi.get_latest_analysis_for_samples_and_type(
                    project=output_project,
                    analysis_type=AnalysisType('cram'),
                    request_body=[s['id']],
                )
                assert cram_analysis, s
            am = AnalysisModel(
                type=AnalysisType(a_type),
                output=f'result.{a_type}',
                status=AnalysisStatus('queued'),
                sample_ids=[s['id']],
            )
            aapi.create_new_analysis(project=output_project, analysis_model=am)
    elif a_type == 'joint-calling':
        for s in samples:
            gvcf_analysis = aapi.get_latest_analysis_for_samples_and_type(
                project=output_project,
                analysis_type=AnalysisType('gvcf'),
                request_body=[s['id']],
            )
            assert gvcf_analysis, s
        am = AnalysisModel(
            sample_ids=[s['id'] for s in samples],
            type=AnalysisType('joint-calling'),
            output='joint-called.vcf',
            status=AnalysisStatus('queued'),
        )
        aapi.create_new_analysis(project=output_project, analysis_model=am)
def _submit_analyses ( samples : List , output_project : str , a_type : str ) : if a_type in [ 'gvcf' , 'cram' ] : for s in samples : if a_type == 'gvcf' : cram_analysis = aapi . get_latest_analysis_for_samples_and_type ( project = output_project , analysis_type = AnalysisType ( 'cram' ) , request_body = [ s [ 'id' ] ] , ) assert cram_analysis , s am = AnalysisModel ( type = AnalysisType ( a_type ) , output = f'result.{a_type}' , status = AnalysisStatus ( 'queued' ) , sample_ids = [ s [ 'id' ] ] , ) aapi . create_new_analysis ( project = output_project , analysis_model = am ) elif a_type == 'joint-calling' : for s in samples : gvcf_analysis = aapi . get_latest_analysis_for_samples_and_type ( project = output_project , analysis_type = AnalysisType ( 'gvcf' ) , request_body = [ s [ 'id' ] ] , ) assert gvcf_analysis , s am = AnalysisModel ( sample_ids = [ s [ 'id' ] for s in samples ] , type = AnalysisType ( 'joint-calling' ) , output = 'joint-called.vcf' , status = AnalysisStatus ( 'queued' ) , ) aapi . create_new_analysis ( project = output_project , analysis_model = am )
Add or update analyses.
Add or update analyses .
"""
    Add or update analyses. Iterate over completed analyses,
    and submit next-step analyses
    """ # completed_analysis = latest_by_type_and_sids.get(('cram', (s['id'],))),
samples
List
output_project
str
a_type
str