def reduce_fit(interface, state, label, inp):
    from disco.util import kvgroup
    import numpy as np
    out = interface.output(0)
    fit_model = {"y_labels": [], "y_sum": 0, "iv": set()}
    combiner = {}
    means, variances = [], []
    k_prev = ""
    for key, value in kvgroup(inp):
        k_split = key.split(state["delimiter"])
        if len(k_split) == 3:
            fit_model["iv"].add(tuple(k_split[1:]))
            out.add(tuple(k_split), sum(value))
        elif len(k_split) == 2:
            if k_split[0] != k_prev and k_prev != "":
                mean, var = zip(*[combiner[key] for key in sorted(combiner.keys())])
                means.append(mean)
                variances.append(var)
            n_a = mean_a = var_a = 0
            for n_b, mean_b, var_b in value:
                n_ab = n_a + n_b
                var_a = ((n_a * var_a + n_b * var_b) / float(n_ab)) + (
                    n_a * n_b * ((mean_b - mean_a) / float(n_ab)) ** 2)
                mean_a = (n_a * mean_a + n_b * mean_b) / float(n_ab)
                n_a = n_ab
            combiner[int(k_split[1])] = (mean_a, var_a + 1e-9)
            k_prev = k_split[0]
        else:
            fit_model[key] = np.sum(value)
            fit_model["y_sum"] += fit_model[key]
            fit_model["y_labels"].append(key)
    if len(means) > 0:
        mean, var = zip(*[combiner[key] for key in sorted(combiner.keys())])
        out.add("mean", np.array(means + [mean], dtype=np.float32))
        variances = np.array(variances + [var], dtype=np.float32)
        out.add("var", variances)
        out.add("var_log", np.log(np.pi * variances))
    prior = [fit_model[y_label] / float(fit_model["y_sum"]) for y_label in fit_model["y_labels"]]
    out.add("prior", np.array(prior, dtype=np.float32))
    out.add("prior_log", np.log(prior))
    out.add("iv", list(fit_model["iv"]))
    out.add("y_labels", fit_model["y_labels"])