def combine_sample_regions(*samples):
    """Create batch-level sets of callable regions for multi-sample calling.

    Intersects all non-callable (nblock) regions from all samples in a batch,
    producing a global set of callable regions.
    """
    samples = utils.unpack_worlds(samples)
    samples = cwlutils.unpack_tarballs(samples, samples[0])
    # back compatibility -- global file for entire sample set
    global_analysis_file = os.path.join(samples[0]["dirs"]["work"], "analysis_blocks.bed")
    if utils.file_exists(global_analysis_file) and not _needs_region_update(global_analysis_file, samples):
        global_no_analysis_file = os.path.join(os.path.dirname(global_analysis_file), "noanalysis_blocks.bed")
    else:
        global_analysis_file = None
    out = []
    analysis_files = []
    batches = []
    with shared.bedtools_tmpdir(samples[0]):
        for batch, items in vmulti.group_by_batch(samples, require_bam=False).items():
            batches.append(items)
            if global_analysis_file:
                analysis_file, no_analysis_file = global_analysis_file, global_no_analysis_file
            else:
                analysis_file, no_analysis_file = _combine_sample_regions_batch(batch, items)
            for data in items:
                vr_file = dd.get_variant_regions(data)
                if analysis_file:
                    analysis_files.append(analysis_file)
                    data["config"]["algorithm"]["callable_regions"] = analysis_file
                    data["config"]["algorithm"]["non_callable_regions"] = no_analysis_file
                    data["config"]["algorithm"]["callable_count"] = pybedtools.BedTool(analysis_file).count()
                elif vr_file:
                    data["config"]["algorithm"]["callable_count"] = pybedtools.BedTool(vr_file).count()
                # attach a representative sample for calculating callable region
                if not data.get("work_bam"):
                    for x in items:
                        if x.get("work_bam"):
                            data["work_bam_callable"] = x["work_bam"]
                out.append([data])
        # Ensure output order matches input order, consistency for CWL-based runs
        assert len(out) == len(samples)
        sample_indexes = {dd.get_sample_name(d): i for i, d in enumerate(samples)}
        def by_input_index(xs):
            return sample_indexes[dd.get_sample_name(xs[0])]
        out.sort(key=by_input_index)
        if len(analysis_files) > 0:
            final_regions = pybedtools.BedTool(analysis_files[0])
            _analysis_block_stats(final_regions, batches[0])
    return out
def combine_sample_regions ( * samples ) : samples = utils . unpack_worlds ( samples ) samples = cwlutils . unpack_tarballs ( samples , samples [ 0 ] ) global_analysis_file = os . path . join ( samples [ 0 ] [ "dirs" ] [ "work" ] , "analysis_blocks.bed" ) if utils . file_exists ( global_analysis_file ) and not _needs_region_update ( global_analysis_file , samples ) : global_no_analysis_file = os . path . join ( os . path . dirname ( global_analysis_file ) , "noanalysis_blocks.bed" ) else : global_analysis_file = None out = [ ] analysis_files = [ ] batches = [ ] with shared . bedtools_tmpdir ( samples [ 0 ] ) : for batch , items in vmulti . group_by_batch ( samples , require_bam = False ) . items ( ) : batches . append ( items ) if global_analysis_file : analysis_file , no_analysis_file = global_analysis_file , global_no_analysis_file else : analysis_file , no_analysis_file = _combine_sample_regions_batch ( batch , items ) for data in items : vr_file = dd . get_variant_regions ( data ) if analysis_file : analysis_files . append ( analysis_file ) data [ "config" ] [ "algorithm" ] [ "callable_regions" ] = analysis_file data [ "config" ] [ "algorithm" ] [ "non_callable_regions" ] = no_analysis_file data [ "config" ] [ "algorithm" ] [ "callable_count" ] = pybedtools . BedTool ( analysis_file ) . count ( ) elif vr_file : data [ "config" ] [ "algorithm" ] [ "callable_count" ] = pybedtools . BedTool ( vr_file ) . count ( ) if not data . get ( "work_bam" ) : for x in items : if x . get ( "work_bam" ) : data [ "work_bam_callable" ] = x [ "work_bam" ] out . append ( [ data ] ) assert len ( out ) == len ( samples ) sample_indexes = { dd . get_sample_name ( d ) : i for i , d in enumerate ( samples ) } def by_input_index ( xs ) : return sample_indexes [ dd . get_sample_name ( xs [ 0 ] ) ] out . sort ( key = by_input_index ) if len ( analysis_files ) > 0 : final_regions = pybedtools . BedTool ( analysis_files [ 0 ] ) _analysis_block_stats ( final_regions , batches [ 0 ] ) return out
Create batch-level sets of callable regions for multi-sample calling.

    Intersects all non-callable (nblock) regions from all samples in a batch,
    producing a global set of callable regions.