Split a set of CWL output dictionaries into data samples and CWL items.

    Handles cases where we're arrayed on multiple things, like a set of regional
    VCF calls and data objects.
348513
def split_data_cwl_items(items, default_keys=None):
    
    key_lens = set([])
    for data in items:
        key_lens.add(len(_get_all_cwlkeys([data], default_keys)))
    extra_key_len = min(list(key_lens)) if len(key_lens) > 1 else None
    data_out = []
    extra_out = []
    for data in items:
        if extra_key_len and len(_get_all_cwlkeys([data], default_keys)) == extra_key_len:
            extra_out.append(data)
        else:
            data_out.append(data)
    if len(extra_out) == 0:
        return data_out, {}
    else:
        cwl_keys = extra_out[0]["cwl_keys"]
        for extra in extra_out[1:]:
            cur_cwl_keys = extra["cwl_keys"]
            assert cur_cwl_keys == cwl_keys, pprint.pformat(extra_out)
        cwl_extras = collections.defaultdict(list)
        for data in items:
            for key in cwl_keys:
                cwl_extras[key].append(data[key])
        data_final = []
        for data in data_out:
            for key in cwl_keys:
                data.pop(key)
            data_final.append(data)
        return data_final, dict(cwl_extras)
732951
def split_sequence_file_on_sample_ids_to_files(seqs,
                                               outdir):
    
    logger = logging.getLogger(__name__)
    logger.info('split_sequence_file_on_sample_ids_to_files'
                ' for file %s into dir %s' % (seqs, outdir))

    outputs = {}

    for bits in sequence_generator(seqs):
        sample = sample_id_from_read_id(bits[0])

        if sample not in outputs:
            outputs[sample] = open(join(outdir, sample + '.fasta'), 'w')
        outputs[sample].write(">%s\n%s\n" % (bits[0], bits[1]))
    for sample in outputs:
        outputs[sample].close()
    logger.info('split to %d files' % len(outputs))
71037
def _split_overlays(self):
        
        if not len(self):
            raise ValueError('Cannot split DynamicMap before it has been initialized')
        elif not issubclass(self.type, CompositeOverlay):
            return None, self

        from ..util import Dynamic
        keys = list(self.last.data.keys())
        dmaps = []
        for key in keys:
            el = self.last.data[key]
            def split_overlay_callback(obj, overlay_key=key, overlay_el=el, **kwargs):
                spec = util.get_overlay_spec(obj, overlay_key, overlay_el)
                items = list(obj.data.items())
                specs = [(i, util.get_overlay_spec(obj, k, v))
                         for i, (k, v) in enumerate(items)]
                match = util.closest_match(spec, specs)
                if match is None:
                    raise KeyError('{spec} spec not found in {otype}. The split_overlays method '
                                   'only works consistently for a DynamicMap where the '
                                   'layers of the {otype} do not change.'.format(
                                       spec=spec, otype=type(obj).__name__))
                return items[match][1]
            dmap = Dynamic(self, streams=self.streams, operation=split_overlay_callback)
            dmap.data = OrderedDict([(list(self.data.keys())[-1], self.last.data[key])])
            dmaps.append(dmap)
        return keys, dmaps
1260411
def _slice_split_info_to_instruction_dicts(self, list_sliced_split_info):
    """"""
    instruction_dicts = []
    for sliced_split_info in list_sliced_split_info:
      mask = splits_lib.slice_to_percent_mask(sliced_split_info.slice_value)

      # Compute filenames from the given split
      filepaths = list(sorted(self._build_split_filenames(
          split_info_list=[sliced_split_info.split_info],
      )))

      # Compute the offsets
      if sliced_split_info.split_info.num_examples:
        shard_id2num_examples = splits_lib.get_shard_id2num_examples(
            sliced_split_info.split_info.num_shards,
            sliced_split_info.split_info.num_examples,
        )
        mask_offsets = splits_lib.compute_mask_offsets(shard_id2num_examples)
      else:
        logging.warning(
            "Statistics not present in the dataset. TFDS is not able to load "
            "the total number of examples, so using the subsplit API may not "
            "provide precise subsplits."
        )
        mask_offsets = [0] * len(filepaths)

      for filepath, mask_offset in zip(filepaths, mask_offsets):
        instruction_dicts.append({
            "filepath": filepath,
            "mask": mask,
            "mask_offset": mask_offset,
        })
    return instruction_dicts
127563
def split(self, start=None, end=None, datatype=None, **kwargs):
        
        if not self.interface.multi:
            if datatype == 'array':
                obj = self.array(**kwargs)
            elif datatype == 'dataframe':
                obj = self.dframe(**kwargs)
            elif datatype == 'columns':
                obj = self.columns(**kwargs)
            elif datatype is None:
                obj = self
            else:
                raise ValueError("%s datatype not support" % datatype)
            return [obj]
        return self.interface.split(self, start, end, datatype, **kwargs)
217924
def merge_split_alignments(samples, run_parallel):
    
    ready = []
    file_key = "work_bam"
    to_merge = collections.defaultdict(list)
    for data in (xs[0] for xs in samples):
        if data.get("combine"):
            out_key = tz.get_in(["combine", file_key, "out"], data)
            if not out_key:
                out_key = data["rgnames"]["lane"]
            to_merge[out_key].append(data)
        else:
            ready.append([data])
    ready_merge = []
    hla_merges = []
    for mgroup in to_merge.values():
        cur_data = mgroup[0]
        del cur_data["align_split"]
        for x in mgroup[1:]:
            cur_data["combine"][file_key]["extras"].append(x[file_key])
        ready_merge.append([cur_data])
        cur_hla = None
        for d in mgroup:
            hla_files = tz.get_in(["hla", "fastq"], d)
            if hla_files:
                if not cur_hla:
                    cur_hla = {"rgnames": {"sample": dd.get_sample_name(cur_data)},
                               "config": cur_data["config"], "dirs": cur_data["dirs"],
                               "hla": {"fastq": []}}
                cur_hla["hla"]["fastq"].append(hla_files)
        if cur_hla:
            hla_merges.append([cur_hla])
    if not tz.get_in(["config", "algorithm", "kraken"], data):
        # kraken requires fasta filenames from data['files'] as input.
        # We don't want to remove those files if kraken qc is required.
        _save_fastq_space(samples)
    merged = run_parallel("delayed_bam_merge", ready_merge)
    hla_merge_raw = run_parallel("merge_split_alignments", hla_merges)
    hla_merges = {}
    for hla_merge in [x[0] for x in hla_merge_raw]:
        hla_merges[dd.get_sample_name(hla_merge)] = tz.get_in(["hla", "fastq"], hla_merge)

    # Add stable 'align_bam' target to use for retrieving raw alignment
    out = []
    for data in [x[0] for x in merged + ready]:
        if data.get("work_bam"):
            data["align_bam"] = data["work_bam"]
        if dd.get_sample_name(data) in hla_merges:
            data["hla"]["fastq"] = hla_merges[dd.get_sample_name(data)]
        else:
            hla_files = glob.glob(os.path.join(dd.get_work_dir(data), "align",
                                               dd.get_sample_name(data), "hla", "*.fq"))
            if hla_files:
                data["hla"]["fastq"] = hla_files
        out.append([data])
    return out
885640
def _split_samples_by_qc(samples):
    
    to_process = []
    extras = []
    for data in [utils.to_single_data(x) for x in samples]:
        qcs = dd.get_algorithm_qc(data)
        # kraken doesn't need bam
        if qcs and (dd.get_align_bam(data) or dd.get_work_bam(data) or
                    tz.get_in(["config", "algorithm", "kraken"], data)):
            for qc in qcs:
                add = copy.deepcopy(data)
                add["config"]["algorithm"]["qc"] = [qc]
                to_process.append([add])
        else:
            extras.append([data])
    return to_process, extras
218075
def _subset_by_support(orig_vcf, cmp_calls, data):
    
    cmp_vcfs = [x["vrn_file"] for x in cmp_calls]
    out_file = "%s-inensemble.vcf.gz" % utils.splitext_plus(orig_vcf)[0]
    if not utils.file_uptodate(out_file, orig_vcf):
        with file_transaction(data, out_file) as tx_out_file:
            cmd = "bedtools intersect -header -wa -f 0.5 -r -a {orig_vcf} -b "
            for cmp_vcf in cmp_vcfs:
                cmd += "<(bcftools view -f 'PASS,.' %s) " % cmp_vcf
            cmd += "| bgzip -c > {tx_out_file}"
            do.run(cmd.format(**locals()), "Subset calls by those present in Ensemble output")
    return vcfutils.bgzip_and_index(out_file, data["config"])
1133682
def splits(self):
        
        return list(map(lambda x: self.select_code(x), self.code))
980200
def combine_calls(*args):
    
    if len(args) == 3:
        is_cwl = False
        batch_id, samples, data = args
        caller_names, vrn_files = _organize_variants(samples, batch_id)
    else:
        is_cwl = True
        samples = [utils.to_single_data(x) for x in args]
        samples = [cwlutils.unpack_tarballs(x, x) for x in samples]
        data = samples[0]
        batch_id = data["batch_id"]
        caller_names = data["variants"]["variantcallers"]
        vrn_files = data["variants"]["calls"]
    logger.info("Ensemble consensus calls for {0}: {1}".format(
        batch_id, ",".join(caller_names)))
    edata = copy.deepcopy(data)
    base_dir = utils.safe_makedir(os.path.join(edata["dirs"]["work"], "ensemble", batch_id))
    if any([vcfutils.vcf_has_variants(f) for f in vrn_files]):
        # Decompose multiallelic variants and normalize
        passonly = not tz.get_in(["config", "algorithm", "ensemble", "use_filtered"], edata, False)
        vrn_files = [normalize.normalize(f, data, passonly=passonly, rerun_effects=False, remove_oldeffects=True,
                                         nonrefonly=True,
                                         work_dir=utils.safe_makedir(os.path.join(base_dir, c)))
                     for c, f in zip(caller_names, vrn_files)]
        if "classifiers" not in (dd.get_ensemble(edata) or {}):
            callinfo = _run_ensemble_intersection(batch_id, vrn_files, caller_names, base_dir, edata)
        else:
            config_file = _write_config_file(batch_id, caller_names, base_dir, edata)
            callinfo = _run_ensemble(batch_id, vrn_files, config_file, base_dir,
                                     dd.get_ref_file(edata), edata)
            callinfo["vrn_file"] = vcfutils.bgzip_and_index(callinfo["vrn_file"], data["config"])
        # After decomposing multiallelic variants and normalizing, re-evaluate effects
        ann_ma_file, _ = effects.add_to_vcf(callinfo["vrn_file"], data)
        if ann_ma_file:
            callinfo["vrn_file"] = ann_ma_file

        edata["config"]["algorithm"]["variantcaller"] = "ensemble"
        edata["vrn_file"] = callinfo["vrn_file"]
        edata["ensemble_bed"] = callinfo["bed_file"]
        callinfo["validate"] = validate.compare_to_rm(edata)[0][0].get("validate")
    else:
        out_vcf_file = os.path.join(base_dir, "{0}-ensemble.vcf".format(batch_id))
        vcfutils.write_empty_vcf(out_vcf_file, samples=[dd.get_sample_name(d) for d in samples])
        callinfo = {"variantcaller": "ensemble",
                    "vrn_file": vcfutils.bgzip_and_index(out_vcf_file, data["config"]),
                    "bed_file": None}
    if is_cwl:
        callinfo["batch_samples"] = data["batch_samples"]
        callinfo["batch_id"] = batch_id
        return [{"ensemble": callinfo}]
    else:
        return [[batch_id, callinfo]]
1045158
def split_variants_by_sample(data):
    
    # not split, do nothing
    if "group_orig" not in data:
        return [[data]]
    # cancer tumor/normal
    elif (vcfutils.get_paired_phenotype(data)
            and "tumor" in [vcfutils.get_paired_phenotype(d) for d in get_orig_items(data)]):
        out = []
        for i, sub_data in enumerate(get_orig_items(data)):
            if vcfutils.get_paired_phenotype(sub_data) == "tumor":
                cur_batch = tz.get_in(["metadata", "batch"], data)
                if cur_batch:
                    sub_data["metadata"]["batch"] = cur_batch
                sub_data["vrn_file"] = data["vrn_file"]
            else:
                sub_data.pop("vrn_file", None)
            out.append([sub_data])
        return out
    # joint calling or population runs, do not split back up and keep in batches
    else:
        out = []
        for sub_data in get_orig_items(data):
            cur_batch = tz.get_in(["metadata", "batch"], data)
            if cur_batch:
                sub_data["metadata"]["batch"] = cur_batch
            sub_data["vrn_file_batch"] = data["vrn_file"]
            sub_data["vrn_file"] = data["vrn_file"]
            out.append([sub_data])
        return out
811437
def samples_to_records(samples, default_keys=None):
    
    from bcbio.pipeline import run_info
    RECORD_CONVERT_TO_LIST = set(["config__algorithm__tools_on", "config__algorithm__tools_off",
                                  "reference__genome_context"])
    all_keys = _get_all_cwlkeys(samples, default_keys)
    out = []
    for data in samples:
        for raw_key in sorted(list(all_keys)):
            key = raw_key.split("__")
            if tz.get_in(key, data) is None:
                data = tz.update_in(data, key, lambda x: None)
            if raw_key not in data["cwl_keys"]:
                data["cwl_keys"].append(raw_key)
            if raw_key in RECORD_CONVERT_TO_LIST:
                val = tz.get_in(key, data)
                if not val: val = []
                elif not isinstance(val, (list, tuple)): val = [val]
                data = tz.update_in(data, key, lambda x: val)
            # Booleans are problematic for CWL serialization, convert into string representation
            if isinstance(tz.get_in(key, data), bool):
                data = tz.update_in(data, key, lambda x: str(tz.get_in(key, data)))
        data["metadata"] = run_info.add_metadata_defaults(data.get("metadata", {}))
        out.append(data)
    return out
370477
def _make_split(self):
        
        self.tile_dict = {}

        wfs = WebFeatureService(self.area_bbox, self.time_interval, data_source=self.data_source,
                                instance_id=self.instance_id)
        date_list = wfs.get_dates()
        geometry_list = wfs.get_geometries()
        for tile_info, (date, geometry) in zip(wfs, zip(date_list, geometry_list)):
            tile_name = ''.join(tile_info['properties']['path'].split('/')[4:7])
            if tile_name not in self.tile_dict:
                self.tile_dict[tile_name] = {'bbox': BBox(tile_info['properties']['mbr'],
                                                          crs=tile_info['properties']['crs']),
                                             'times': [],
                                             'geometries': []}
            self.tile_dict[tile_name]['times'].append(date)
            self.tile_dict[tile_name]['geometries'].append(geometry)

        self.tile_dict = {tile_name: tile_props for tile_name, tile_props in self.tile_dict.items() if
                          self._intersects_area(tile_props['bbox'])}

        self.bbox_list = []
        self.info_list = []

        for tile_name, tile_info in self.tile_dict.items():
            tile_bbox = tile_info['bbox']
            bbox_splitter = BBoxSplitter([tile_bbox.geometry], tile_bbox.crs,
                                         split_shape=self.tile_split_shape)

            for bbox, info in zip(bbox_splitter.get_bbox_list(), bbox_splitter.get_info_list()):
                if self._intersects_area(bbox):
                    info['tile'] = tile_name

                    self.bbox_list.append(bbox)
                    self.info_list.append(info)
435581
def dataset_splits(self):
    """"""
    return [{
        "split": problem.DatasetSplit.TRAIN,
        "shards": _TRAIN_SHARDS,
    }, {
        "split": problem.DatasetSplit.EVAL,
        "shards": _DEV_SHARDS,
    }]
1254890
def ParseMultiple(self, result_dicts):
    """"""
    for result_dict in result_dicts:
      result = result_dict.ToDict()
      winvolume = rdf_client_fs.WindowsVolume(
          drive_letter=result.get("DeviceID"),
          drive_type=result.get("DriveType"))

      try:
        size = int(result.get("Size"))
      except (ValueError, TypeError):
        size = None

      try:
        free_space = int(result.get("FreeSpace"))
      except (ValueError, TypeError):
        free_space = None

      # Since we don't get the sector sizes from WMI, we just set them at 1 byte
      yield rdf_client_fs.Volume(
          windowsvolume=winvolume,
          name=result.get("VolumeName"),
          file_system_type=result.get("FileSystem"),
          serial_number=result.get("VolumeSerialNumber"),
          sectors_per_allocation_unit=1,
          bytes_per_sector=1,
          total_allocation_units=size,
          actual_available_allocation_units=free_space)
520143
def dataset_splits(self):
    """"""
    return [{
        "split": problem.DatasetSplit.TRAIN,
        "shards": self.num_train_shards,
    }, {
        "split": problem.DatasetSplit.EVAL,
        "shards": self.num_eval_shards,
    }, {
        "split": problem.DatasetSplit.TEST,
        "shards": self.num_test_shards,
    }]
779014
def _split_diff(merge_result, context_lines=3):
    """"""
    collect = []
    for item in _visible_in_diff(merge_result, context_lines=context_lines):
        if item is None:
            if collect:
                yield collect
            collect = []
        else:
            collect.append(item)