def _merge_hla_fastq_inputs(data):
    """Merge HLA inputs from a split initial alignment.
    """
    hla_key = ["hla", "fastq"]
    hla_sample_files = [x for x in tz.get_in(hla_key, data, []) if x and x != "None"]
    if hla_sample_files:
        out_files = collections.defaultdict(list)
        for hla_files in hla_sample_files:
            for hla_file in hla_files:
                rehla = re.search(".hla.(?P<hlatype>[\w-]+).fq", hla_file)
                if rehla:
                    hlatype = rehla.group("hlatype")
                    out_files[hlatype].append(hla_file)
        if len(out_files) > 0:
            hla_outdir = utils.safe_makedir(os.path.join(dd.get_work_dir(data), "align",
                                                         dd.get_sample_name(data), "hla"))
            merged_hlas = []
            for hlatype, files in out_files.items():
                out_file = os.path.join(hla_outdir, "%s-%s.fq" % (dd.get_sample_name(data), hlatype))
                optitype.combine_hla_fqs([(hlatype, f) for f in files], out_file, data)
                merged_hlas.append(out_file)
        data = tz.update_in(data, hla_key, lambda x: merged_hlas)
    else:
        data = tz.update_in(data, hla_key, lambda x: None)
    return data

def samples_to_records(samples, default_keys=None):
    """Convert samples into output CWL records.
    """
    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

def batch_for_variantcall(samples):
    """Prepare a set of samples for parallel variant calling.

    CWL input target that groups samples into batches and variant callers
    for parallel processing.
    """
    convert_to_list = set(["config__algorithm__tools_on", "config__algorithm__tools_off"])
    to_process, extras = _dup_samples_by_variantcaller(samples, require_bam=False)
    batch_groups = collections.defaultdict(list)
    to_process = [utils.to_single_data(x) for x in to_process]
    all_keys = set([])
    for data in to_process:
        all_keys.update(set(data["cwl_keys"]))
    for data in to_process:
        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)
                data["cwl_keys"].append(raw_key)
            if raw_key in 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)
        vc = get_variantcaller(data, require_bam=False)
        batches = dd.get_batches(data) or dd.get_sample_name(data)
        if not isinstance(batches, (list, tuple)):
            batches = [batches]
        for b in batches:
            batch_groups[(b, vc)].append(utils.deepish_copy(data))
    return list(batch_groups.values()) + extras

def run_and_save(data):
    """Run QC, saving file outputs in data dictionary.
    """
    run(None, data)
    stats_file, idxstats_file = _get_stats_files(data)
    data = tz.update_in(data, ["depth", "samtools", "stats"], lambda x: stats_file)
    data = tz.update_in(data, ["depth", "samtools", "idxstats"], lambda x: idxstats_file)
    return data

 def _compare_dicts(self, orig, new, ns):
     out = {}
     for key, val in new.items():
         nskey = ns + [key]
         orig_val = tz.get_in([key], orig)
         if isinstance(val, dict) and isinstance(orig_val, dict):
             for nkey, nval in self._compare_dicts(orig_val or {}, val or {}, nskey).items():
                 out = tz.update_in(out, [nkey], lambda x: nval)
         elif val != orig_val:
             print nskey, val, orig_val
             out = tz.update_in(out, nskey, lambda x: val)
     return out

def assign_complex_to_samples(items):
    """Assign complex inputs like variants and align outputs to samples.

    Handles list inputs to record conversion where we have inputs from multiple
    locations and need to ensure they are properly assigned to samples in many
    environments.

    The unpleasant approach here is to use standard file naming to match
    with samples so this can work in environments where we don't download/stream
    the input files (for space/time savings).
    """
    extract_fns = {("variants", "samples"): _get_vcf_samples,
                   ("align_bam",): _get_bam_samples}
    complex = {k: {} for k in extract_fns.keys()}
    for data in items:
        for k in complex:
            v = tz.get_in(k, data)
            if v is not None:
                for s in extract_fns[k](v, items):
                    if s:
                        complex[k][s] = v
    out = []
    for data in items:
        for k in complex:
            newv = tz.get_in([k, dd.get_sample_name(data)], complex)
            if newv:
                data = tz.update_in(data, k, lambda x: newv)
        out.append(data)
    return out

def calculate_sv_coverage(data):
    """Calculate coverage within bins for downstream CNV calling.

    Creates corrected cnr files with log2 ratios and depths.
    """
    calcfns = {"cnvkit": _calculate_sv_coverage_cnvkit, "gatk-cnv": _calculate_sv_coverage_gatk}
    from bcbio.structural import cnvkit
    data = utils.to_single_data(data)
    if not cnvkit.use_general_sv_bins(data):
        out_target_file, out_anti_file = (None, None)
    else:
        work_dir = utils.safe_makedir(os.path.join(dd.get_work_dir(data), "structural",
                                                   dd.get_sample_name(data), "bins"))
        out_target_file, out_anti_file = calcfns[cnvkit.bin_approach(data)](data, work_dir)
        if not os.path.exists(out_target_file):
            out_target_file, out_anti_file = (None, None)
    if "seq2c" in dd.get_svcaller(data):
        from bcbio.structural import seq2c
        seq2c_target = seq2c.precall(data)
    else:
        seq2c_target = None

    if not tz.get_in(["depth", "bins"], data):
        data = tz.update_in(data, ["depth", "bins"], lambda x: {})
    data["depth"]["bins"] = {"target": out_target_file, "antitarget": out_anti_file, "seq2c": seq2c_target}
    return [[data]]

def vc_output_record(samples):
    """Prepare output record from variant calling to feed into downstream analysis.

    Prep work handles reformatting so we return generated dictionaries.

    For any shared keys that are calculated only once for a batch, like variant calls
    for the batch, we assign to every sample.
    """
    shared_keys = [["vrn_file"], ["validate", "summary"],
                   ["validate", "tp"], ["validate", "fp"], ["validate", "fn"]]
    raw = cwlutils.samples_to_records([utils.to_single_data(x) for x in samples])
    shared = {}
    for key in shared_keys:
        cur = list(set([x for x in [tz.get_in(key, d) for d in raw] if x]))
        if len(cur) > 0:
            assert len(cur) == 1, (key, cur)
            shared[tuple(key)] = cur[0]
        else:
            shared[tuple(key)] = None
    out = []
    for d in raw:
        for key, val in shared.items():
            d = tz.update_in(d, key, lambda x: val)
        out.append([d])
    return out

def _place_secondary_files(inp_tool, inp_binding):
    """Put secondaryFiles at the level of the File item to ensure indexes get passed.
    """
    secondary_files = inp_tool.pop("secondaryFiles", None)
    if secondary_files:
        key = []
        while tz.get_in(key + ["type"], inp_tool) != "File" and tz.get_in(key + ["items"], inp_tool) != "File":
            key.append("type")
        if tz.get_in(key, inp_tool):
            inp_tool = tz.update_in(inp_tool, key + ["secondaryFiles"], lambda x: secondary_files)
        else:
            nested_inp_binding = copy.deepcopy(inp_binding)
            nested_inp_binding["prefix"] = "ignore="
            nested_inp_binding["secondaryFiles"] = secondary_files
            inp_tool = tz.update_in(inp_tool, key, lambda x: nested_inp_binding)
    return inp_tool

def _fill_prioritization_targets(data):
    """Fill in globally installed files for prioritization.
    """
    ref_file = dd.get_ref_file(data)
    for target in [["svprioritize"]]:
        val = tz.get_in(["config", "algorithm"] + target, data)
        if val and not os.path.exists(val):
            installed_vals = glob.glob(os.path.normpath(os.path.join(os.path.dirname(ref_file), os.pardir,
                                                                     "coverage", "prioritize", val + "*.bed.gz")))
            if len(installed_vals) == 0:
                raise ValueError("Configuration problem. Prioritization file not found for %s: %s" %
                                 (target, val))
            elif len(installed_vals) == 1:
                installed_val = installed_vals[0]
            else:
                # check for partial matches
                installed_val = None
                for v in installed_vals:
                    if v.endswith(val + ".bed.gz"):
                        installed_val = v
                        break
                # handle date-stamped inputs
                if not installed_val:
                    installed_val = sorted(installed_vals, reverse=True)[0]
            data = tz.update_in(data, ["config", "algorithm"] + target, lambda x: installed_val)
    return data

def _fill_prioritization_targets(data):
    """Fill in globally installed files for prioritization.
    """
    ref_file = dd.get_ref_file(data)
    for target in [["svprioritize"], ["coverage"]]:
        val = tz.get_in(["config", "algorithm"] + target, data)
        if val and not os.path.exists(val):
            installed_vals = []
            # Check prioritize directory
            for ext in [".bed", ".bed.gz"]:
                installed_vals += glob.glob(os.path.normpath(os.path.join(os.path.dirname(ref_file), os.pardir,
                                                                          "coverage", "prioritize",
                                                                          val + "*%s" % ext)))
            # Check sv-annotation directory for prioritize gene name lists
            if target[-1] == "svprioritize":
                installed_vals += glob.glob(os.path.join(
                    os.path.dirname(os.path.realpath(utils.which("simple_sv_annotation.py"))),
                    "%s*" % os.path.basename(val)))
            if len(installed_vals) == 0:
                raise ValueError("Configuration problem. BED file not found for %s: %s" %
                                 (target, val))
            elif len(installed_vals) == 1:
                installed_val = installed_vals[0]
            else:
                # check for partial matches
                installed_val = None
                for v in installed_vals:
                    if v.endswith(val + ".bed.gz") or v.endswith(val + ".bed"):
                        installed_val = v
                        break
                # handle date-stamped inputs
                if not installed_val:
                    installed_val = sorted(installed_vals, reverse=True)[0]
            data = tz.update_in(data, ["config", "algorithm"] + target, lambda x: installed_val)
    return data

def _merge_dreams(first, second):
    new_dask = second.dask
    for i in range(second.npartitions):
        # Populate root of first dask into empty of second
        empty_path = dfs_first_empty_path(new_dask)
        new_dask = toolz.update_in(new_dask, empty_path,
                                   lambda _: (first.name, i))
    new_dask = toolz.merge(new_dask, first.dask)
    return type(second)(new_dask, second.name, second.npartitions)

def json_expand(json_op, key_name='json'):
    """ Convert a string json object to Python dict in an op. """
    if type(json_op) == dict and key_name in json_op and json_op[key_name]:
        try:
            return update_in(json_op, [key_name], json.loads)
        except JSONDecodeError:
            return assoc(json_op, key_name, {})

    return json_op

    def transform_network(self, network):
        def update_fn(override_hyperparameters):
            return toolz.merge(override_hyperparameters,
                               self.hyperparameters)

        kwargs = toolz.update_in(transforms.fns.network_to_kwargs(network),
                                 ["override_hyperparameters"],
                                 update_fn)
        return treeano.Network(**kwargs)

def _update_nested(key, val, data):
    """Update the data object, avoiding over-writing with nested dictionaries.
    """
    if isinstance(val, dict):
        for sub_key, sub_val in val.items():
            data = _update_nested(key + [sub_key], sub_val, data)
    else:
        if tz.get_in(key, data) is not None:
            raise ValueError("Duplicated key %s" % key)
        data = tz.update_in(data, key, lambda x: val)
    return data

 def _compare_dicts(self, orig, new, ns):
     out = {}
     for key, val in new.items():
         nskey = ns + [key]
         orig_val = tz.get_in([key], orig)
         if isinstance(val, dict) and isinstance(orig_val, dict):
             for nkey, nval in self._compare_dicts(orig_val or {}, val or {}, nskey).items():
                 out = self._merge(out, {nkey: nval})
         elif val != orig_val:
             out = tz.update_in(out, nskey, lambda x: copy.deepcopy(val))
     return out

def _merge_align_bams(data):
    """Merge multiple alignment BAMs, including split and discordant reads.
    """
    for key in (["work_bam"], ["work_bam_plus", "disc"], ["work_bam_plus", "sr"]):
        in_files = tz.get_in(key, data, [])
        if not isinstance(in_files, (list, tuple)):
            in_files = [in_files]
        in_files = [x for x in in_files if x and x != "None"]
        if in_files:
            ext = "-%s" % key[-1] if len(key) > 1 else ""
            out_file = os.path.join(dd.get_work_dir(data), "align", dd.get_sample_name(data),
                                    "%s-sort%s.bam" % (dd.get_sample_name(data), ext))
            merged_file = merge_bam_files(in_files, utils.safe_makedir(os.path.dirname(out_file)),
                                          data, out_file=out_file)
            data = tz.update_in(data, key, lambda x: merged_file)
        else:
            data = tz.update_in(data, key, lambda x: None)
    if "align_bam" in data and "work_bam" in data:
        data["align_bam"] = data["work_bam"]
    return data

def _symlink_to_workdir(data, key):
    """For CWL support, symlink files into a working directory if in read-only imports.
    """
    orig_file = tz.get_in(key, data)
    if orig_file and not orig_file.startswith(dd.get_work_dir(data)):
        variantcaller = genotype.get_variantcaller(data)
        out_file = os.path.join(dd.get_work_dir(data), variantcaller, os.path.basename(orig_file))
        utils.safe_makedir(os.path.dirname(out_file))
        utils.symlink_plus(orig_file, out_file)
        data = tz.update_in(data, key, lambda x: out_file)
    return data

def _place_secondary_files(inp_tool, inp_binding):
    """Put secondaryFiles at the level of the File item to ensure indexes get passed.

    This involves using a second input binding to get the secondaryFiles, that
    we ignore downstream. Ideally we could use `valueFrom: null` but that doesn't
    seem to work right now.
    """
    secondary_files = inp_tool.pop("secondaryFiles", None)
    if secondary_files:
        key = []
        while tz.get_in(key + ["type"], inp_tool) != "File" and tz.get_in(key + ["items"], inp_tool) != "File":
            key.append("type")
        secondary_key = key + ["inputBinding"]
        if tz.get_in(secondary_key, inp_tool):
            inp_tool = tz.update_in(inp_tool, secondary_key + ["secondaryFiles"], lambda x: secondary_files)
        else:
            nested_inp_binding = copy.deepcopy(inp_binding)
            nested_inp_binding["prefix"] = "ignore="
            nested_inp_binding["secondaryFiles"] = secondary_files
            inp_tool = tz.update_in(inp_tool, secondary_key, lambda x: nested_inp_binding)
    return inp_tool

def _samples_to_records(samples):
    """Convert samples into output CWL records.
    """
    from bcbio.pipeline import run_info
    RECORD_CONVERT_TO_LIST = set(["config__algorithm__tools_on", "config__algorithm__tools_off"])
    all_keys = _get_all_cwlkeys(samples)
    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)
                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)
        data["metadata"] = run_info.add_metadata_defaults(data.get("metadata", {}))
        out.append(data)
    return out