scripts/create_test_subset.py
def main(
    project: str,
    samples_n: Optional[int],
    families_n: Optional[int],
):
    """
    Script creates a test subset for a given project.
    A new project with a prefix -test is created, and for any files in sample/meta,
    sequence/meta, or analysis/output a copy in the -test namespace is created.
    """
    samples_n, families_n = _validate_opts(samples_n, families_n)

    all_samples = sapi.get_samples(
        body_get_samples_by_criteria_api_v1_sample_post={
            'project_ids': [project],
            'active': True,
        }
    )
    logger.info(f'Found {len(all_samples)} samples')
    if samples_n and samples_n >= len(all_samples):
        resp = str(
            input(
                f'Requesting {samples_n} samples which is >= '
                f'than the number of available samples ({len(all_samples)}). '
                f'The test project will be a copy of the production project. '
                f'Please confirm (y): '
            )
        )
        if resp.lower() != 'y':
            raise SystemExit()

    random.seed(42)  # for reproducibility

    pid_sid = papi.get_external_participant_id_to_internal_sample_id(project)
    sample_id_by_participant_id = dict(pid_sid)

    ped_lines = export_ped_file(project, replace_with_participant_external_ids=True)
    if families_n is not None:
        ped = Ped(ped_lines)
        families = list(ped.families.values())
        logger.info(f'Found {len(families)} families, by size:')
        _print_fam_stats(families)
        families = random.sample(families, families_n)
        logger.info(f'After subsetting to {len(families)} families:')
        _print_fam_stats(families)
        sample_ids = []
        for fam in families:
            for s in fam.samples:
                sample_ids.append(sample_id_by_participant_id[s.sample_id])
        samples = [s for s in all_samples if s['id'] in sample_ids]

    else:
        assert samples_n
        samples = random.sample(all_samples, samples_n)
        sample_ids = [s['id'] for s in samples]

    logger.info(
        f'Subset to {len(samples)} samples (internal ID / external ID): '
        f'{_pretty_format_samples(samples)}'
    )

    # Populating test project
    target_project = project + '-test'
    logger.info('Checking any existing test samples in the target test project')

    test_sample_by_external_id = _process_existing_test_samples(target_project, samples)

    try:
        seq_infos: List[Dict] = seqapi.get_sequences_by_sample_ids(sample_ids)
    except exceptions.ApiException:
        seq_info_by_s_id = {}
    else:
        seq_info_by_s_id = dict(zip(sample_ids, seq_infos))

    analysis_by_sid_by_type: Dict[str, Dict] = {'cram': {}, 'gvcf': {}}
    for a_type, analysis_by_sid in analysis_by_sid_by_type.items():
        try:
            analyses: List[Dict] = aapi.get_latest_analysis_for_samples_and_type(
                project=project,
                analysis_type=AnalysisType(a_type),
                request_body=sample_ids,
            )
        except exceptions.ApiException:
            traceback.print_exc()
        else:
            for a in analyses:
                analysis_by_sid[a['sample_ids'][0]] = a
        logger.info(f'Will copy {a_type} analysis entries: {analysis_by_sid}')

    for s in samples:
        logger.info(f'Processing sample {s["id"]}')

        if s['external_id'] in test_sample_by_external_id:
            new_s_id = test_sample_by_external_id[s['external_id']]['id']
            logger.info(f'Sample already in test project, with ID {new_s_id}')
        else:
            logger.info('Creating test sample entry')
            new_s_id = sapi.create_new_sample(
                project=target_project,
                new_sample=NewSample(
                    external_id=s['external_id'],
                    type=s['type'],
                    meta=_copy_files_in_dict(s['meta'], project),
                ),
            )
            seq_info = seq_info_by_s_id.get(s['id'])
            if seq_info:
                logger.info('Processing sequence entry')
                new_meta = _copy_files_in_dict(seq_info.get('meta'), project)
                logger.info('Creating sequence entry in test')
                seqapi.create_new_sequence(
                    new_sequence=NewSequence(
                        sample_id=new_s_id,
                        meta=new_meta,
                        type=seq_info['type'],
                        status=seq_info['status'],
                    )
                )

        for a_type in ['cram', 'gvcf']:
            analysis = analysis_by_sid_by_type[a_type].get(s['id'])
            if analysis:
                logger.info(f'Processing {a_type} analysis entry')
                am = AnalysisModel(
                    type=a_type,
                    output=_copy_files_in_dict(analysis['output'], project),
                    status=analysis['status'],
                    sample_ids=[s['id']],
                )
                logger.info(f'Creating {a_type} analysis entry in test')
                aapi.create_new_analysis(project=target_project, analysis_model=am)
        logger.info(f'-')

    Script creates a test subset for a given project.
    A new project with a prefix -test is created, and for any files in sample/meta,
    sequence/meta, or analysis/output a copy in the -test namespace is created.
    
Script creates a test subset for a given project.
A new project with a prefix -test is created, and for any files in sample/meta,
sequence/meta, or analysis/output a copy in the -test namespace is created.
Script creates a test subset for a given project . A new project with a prefix - test is created and for any files in sample / meta sequence / meta or analysis / output a copy in the - test namespace is created .
def main(
    project: str,
    samples_n: Optional[int],
    families_n: Optional[int],
):
    samples_n, families_n = _validate_opts(samples_n, families_n)
    all_samples = sapi.get_samples(
        body_get_samples_by_criteria_api_v1_sample_post={
            'project_ids': [project],
            'active': True,
        }
    )
    logger.info(f'Found {len(all_samples)} samples')
    if samples_n and samples_n >= len(all_samples):
        resp = str(
            input(
                f'Requesting {samples_n} samples which is >= '
                f'than the number of available samples ({len(all_samples)}). '
                f'The test project will be a copy of the production project. '
                f'Please confirm (y): '
            )
        )
        if resp.lower() != 'y':
            raise SystemExit()
    random.seed(42)  
    pid_sid = papi.get_external_participant_id_to_internal_sample_id(project)
    sample_id_by_participant_id = dict(pid_sid)
    ped_lines = export_ped_file(project, replace_with_participant_external_ids=True)
    if families_n is not None:
        ped = Ped(ped_lines)
        families = list(ped.families.values())
        logger.info(f'Found {len(families)} families, by size:')
        _print_fam_stats(families)
        families = random.sample(families, families_n)
        logger.info(f'After subsetting to {len(families)} families:')
        _print_fam_stats(families)
        sample_ids = []
        for fam in families:
            for s in fam.samples:
                sample_ids.append(sample_id_by_participant_id[s.sample_id])
        samples = [s for s in all_samples if s['id'] in sample_ids]
    else:
        assert samples_n
        samples = random.sample(all_samples, samples_n)
        sample_ids = [s['id'] for s in samples]
    logger.info(
        f'Subset to {len(samples)} samples (internal ID / external ID): '
        f'{_pretty_format_samples(samples)}'
    )
    target_project = project + '-test'
    logger.info('Checking any existing test samples in the target test project')
    test_sample_by_external_id = _process_existing_test_samples(target_project, samples)
    try:
        seq_infos: List[Dict] = seqapi.get_sequences_by_sample_ids(sample_ids)
    except exceptions.ApiException:
        seq_info_by_s_id = {}
    else:
        seq_info_by_s_id = dict(zip(sample_ids, seq_infos))
    analysis_by_sid_by_type: Dict[str, Dict] = {'cram': {}, 'gvcf': {}}
    for a_type, analysis_by_sid in analysis_by_sid_by_type.items():
        try:
            analyses: List[Dict] = aapi.get_latest_analysis_for_samples_and_type(
                project=project,
                analysis_type=AnalysisType(a_type),
                request_body=sample_ids,
            )
        except exceptions.ApiException:
            traceback.print_exc()
        else:
            for a in analyses:
                analysis_by_sid[a['sample_ids'][0]] = a
        logger.info(f'Will copy {a_type} analysis entries: {analysis_by_sid}')
    for s in samples:
        logger.info(f'Processing sample {s["id"]}')
        if s['external_id'] in test_sample_by_external_id:
            new_s_id = test_sample_by_external_id[s['external_id']]['id']
            logger.info(f'Sample already in test project, with ID {new_s_id}')
        else:
            logger.info('Creating test sample entry')
            new_s_id = sapi.create_new_sample(
                project=target_project,
                new_sample=NewSample(
                    external_id=s['external_id'],
                    type=s['type'],
                    meta=_copy_files_in_dict(s['meta'], project),
                ),
            )
            seq_info = seq_info_by_s_id.get(s['id'])
            if seq_info:
                logger.info('Processing sequence entry')
                new_meta = _copy_files_in_dict(seq_info.get('meta'), project)
                logger.info('Creating sequence entry in test')
                seqapi.create_new_sequence(
                    new_sequence=NewSequence(
                        sample_id=new_s_id,
                        meta=new_meta,
                        type=seq_info['type'],
                        status=seq_info['status'],
                    )
                )
        for a_type in ['cram', 'gvcf']:
            analysis = analysis_by_sid_by_type[a_type].get(s['id'])
            if analysis:
                logger.info(f'Processing {a_type} analysis entry')
                am = AnalysisModel(
                    type=a_type,
                    output=_copy_files_in_dict(analysis['output'], project),
                    status=analysis['status'],
                    sample_ids=[s['id']],
                )
                logger.info(f'Creating {a_type} analysis entry in test')
                aapi.create_new_analysis(project=target_project, analysis_model=am)
        logger.info(f'-')
def main ( project : str , samples_n : Optional [ int ] , families_n : Optional [ int ] , ) : samples_n , families_n = _validate_opts ( samples_n , families_n ) all_samples = sapi . get_samples ( body_get_samples_by_criteria_api_v1_sample_post = { 'project_ids' : [ project ] , 'active' : True , } ) logger . info ( f'Found {len(all_samples)} samples' ) if samples_n and samples_n >= len ( all_samples ) : resp = str ( input ( f'Requesting {samples_n} samples which is >= ' f'than the number of available samples ({len(all_samples)}). ' f'The test project will be a copy of the production project. ' f'Please confirm (y): ' ) ) if resp . lower ( ) != 'y' : raise SystemExit ( ) random . seed ( 42 ) pid_sid = papi . get_external_participant_id_to_internal_sample_id ( project ) sample_id_by_participant_id = dict ( pid_sid ) ped_lines = export_ped_file ( project , replace_with_participant_external_ids = True ) if families_n is not None : ped = Ped ( ped_lines ) families = list ( ped . families . values ( ) ) logger . info ( f'Found {len(families)} families, by size:' ) _print_fam_stats ( families ) families = random . sample ( families , families_n ) logger . info ( f'After subsetting to {len(families)} families:' ) _print_fam_stats ( families ) sample_ids = [ ] for fam in families : for s in fam . samples : sample_ids . append ( sample_id_by_participant_id [ s . sample_id ] ) samples = [ s for s in all_samples if s [ 'id' ] in sample_ids ] else : assert samples_n samples = random . sample ( all_samples , samples_n ) sample_ids = [ s [ 'id' ] for s in samples ] logger . info ( f'Subset to {len(samples)} samples (internal ID / external ID): ' f'{_pretty_format_samples(samples)}' ) target_project = project + '-test' logger . info ( 'Checking any existing test samples in the target test project' ) test_sample_by_external_id = _process_existing_test_samples ( target_project , samples ) try : seq_infos : List [ Dict ] = seqapi . get_sequences_by_sample_ids ( sample_ids ) except exceptions . ApiException : seq_info_by_s_id = { } else : seq_info_by_s_id = dict ( zip ( sample_ids , seq_infos ) ) analysis_by_sid_by_type : Dict [ str , Dict ] = { 'cram' : { } , 'gvcf' : { } } for a_type , analysis_by_sid in analysis_by_sid_by_type . items ( ) : try : analyses : List [ Dict ] = aapi . get_latest_analysis_for_samples_and_type ( project = project , analysis_type = AnalysisType ( a_type ) , request_body = sample_ids , ) except exceptions . ApiException : traceback . print_exc ( ) else : for a in analyses : analysis_by_sid [ a [ 'sample_ids' ] [ 0 ] ] = a logger . info ( f'Will copy {a_type} analysis entries: {analysis_by_sid}' ) for s in samples : logger . info ( f'Processing sample {s["id"]}' ) if s [ 'external_id' ] in test_sample_by_external_id : new_s_id = test_sample_by_external_id [ s [ 'external_id' ] ] [ 'id' ] logger . info ( f'Sample already in test project, with ID {new_s_id}' ) else : logger . info ( 'Creating test sample entry' ) new_s_id = sapi . create_new_sample ( project = target_project , new_sample = NewSample ( external_id = s [ 'external_id' ] , type = s [ 'type' ] , meta = _copy_files_in_dict ( s [ 'meta' ] , project ) , ) , ) seq_info = seq_info_by_s_id . get ( s [ 'id' ] ) if seq_info : logger . info ( 'Processing sequence entry' ) new_meta = _copy_files_in_dict ( seq_info . get ( 'meta' ) , project ) logger . info ( 'Creating sequence entry in test' ) seqapi . create_new_sequence ( new_sequence = NewSequence ( sample_id = new_s_id , meta = new_meta , type = seq_info [ 'type' ] , status = seq_info [ 'status' ] , ) ) for a_type in [ 'cram' , 'gvcf' ] : analysis = analysis_by_sid_by_type [ a_type ] . get ( s [ 'id' ] ) if analysis : logger . info ( f'Processing {a_type} analysis entry' ) am = AnalysisModel ( type = a_type , output = _copy_files_in_dict ( analysis [ 'output' ] , project ) , status = analysis [ 'status' ] , sample_ids = [ s [ 'id' ] ] , ) logger . info ( f'Creating {a_type} analysis entry in test' ) aapi . create_new_analysis ( project = target_project , analysis_model = am ) logger . info ( f'-' )
Script creates a test subset for a given project.
Script creates a test subset for a given project .
"""
    Script creates a test subset for a given project.
    A new project with a prefix -test is created, and for any files in sample/meta,
    sequence/meta, or analysis/output a copy in the -test namespace is created.
    """ # for reproducibility # Populating test project
project
str
samples_n
Optional[int]
families_n
Optional[int]