在略有不同的函数中删除重复代码

Question

我有两个非常相似的循环，这两个包含一个与第三个循环非常相似的内部循环（嗯...:)）。用代码说明它看起来接近这个：

# First function
def fmeasure_kfold1(array, nfolds):
    ret = []

    # Kfold1 and kfold2 both have this outer loop
    for train_index, test_index in KFold(len(array), nfolds):
        correlation = analyze(array[train_index])

        for build in array[test_index]:  # <- All functions have this loop

            # Retrieved tests is calculated inside the build loop in kfold1
            retrieved_tests = get_tests(set(build['modules']), correlation)

            relevant_tests = set(build['tests'])
            fval = calc_f(relevant_tests, retrieved_tests)
            if fval is not None:
                ret.append(fval)

    return ret

# Second function
def fmeasure_kfold2(array, nfolds):
    ret = []

    # Kfold1 and kfold2 both have this outer loop
    for train_index, test_index in KFold(len(array), nfolds):
        correlation = analyze(array[train_index])

        # Retrieved tests is calculated outside the build loop in kfold2
        retrieved_tests = _sum_tests(correlation)

        for build in array[test_index]:  # <- All functions have this loop

            relevant_tests = set(build['tests'])
            fval = calc_f(relevant_tests, retrieved_tests)
            if fval is not None:
                ret.append(fval)

    return ret

# Third function
def fmeasure_all(array):
    ret = []
    for build in array:  # <- All functions have this loop

        relevant = set(build['tests'])
        fval = calc_f2(relevant)  # <- Instead of calc_f, I call calc_f2
        if fval is not None:
            ret.append(fval)

    return ret

前两个函数只是方式不同，在什么时候计算retrieved_tests。第三个函数与前两个函数的内部循环不同，它调用calc_f2，并且不使用retrieved_tests。

实际上代码更复杂，但虽然重复让我很恼火，但我想我可以忍受它。不过最近一直在改，一次改两三个地方很烦。

如何合并重复的代码？我能想到的唯一方法是引入类，这引入了很多样板文件，如果可能的话，我希望将函数保留为纯函数。

---

编辑

这是calc_f和calc_f2的内容：

def calc_f(relevant, retrieved):
    """Calculate the F-measure given relevant and retrieved tests."""
    recall = len(relevant & retrieved)/len(relevant)
    prec = len(relevant & retrieved)/len(retrieved)
    fmeasure = f_measure(recall, prec)

    return (fmeasure, recall, prec)


def calc_f2(relevant, nbr_tests=1000):
    """Calculate the F-measure given relevant tests."""
    recall = 1
    prec = len(relevant) / nbr_tests
    fmeasure = f_measure(recall, prec)

    return (fmeasure, recall, prec)

f_measure 计算准确率和召回率的harmonic mean。

基本上，calc_f2 需要很多捷径，因为不需要检索到的测试。

Answer 1

拥有一个带有额外参数的通用函数来控制retrieved_tests 的计算位置也可以。

例如

def fmeasure_kfold_generic(array, nfolds, mode):
    ret = []

    # Kfold1 and kfold2 both have this outer loop
    for train_index, test_index in KFold(len(array), nfolds):
        correlation = analyze(array[train_index])

        # Retrieved tests is calculated outside the build loop in kfold2
        if mode==2:
            retrieved_tests = _sum_tests(correlation)

        for build in array[test_index]:  # <- All functions have this loop
            # Retrieved tests is calculated inside the build loop in kfold1
            if mode==1:
                retrieved_tests = get_tests(set(build['modules']), correlation)

            relevant_tests = set(build['tests'])
            fval = calc_f(relevant_tests, retrieved_tests)
            if fval is not None:
                ret.append(fval)

Answer 2

一种方法是将每个内部循环编写为一个函数，然后将外部循环作为一个单独的函数，接收其他循环作为参数。这类似于排序函数（接收应该用于比较两个元素的函数）中所做的事情。

当然，困难的部分是找出所有函数之间的共同部分到底是什么，这并不总是那么简单。

Answer 3

典型的解决方案是识别算法的各个部分并使用Template method design pattern，其中不同的阶段将在子类中实现。我根本看不懂你的代码，但我认为会有makeGlobalRetrievedTests() 和makeIndividualRetrievedTests() 之类的方法？

Answer 4

我会从里到外解决问题：通过分解最里面的循环。这适用于“函数式”风格（以及“函数式编程”）。在我看来，如果您将fmeasure_all 概括一下，就可以实现所有三个功能。类似的东西

def fmeasure(builds, calcFn, retrieveFn):
    ret = []
    for build in array:
        relevant = set(build['tests'])
        fval = calcFn(relevant, retrieveFn(build))
        if fval is not None:
            ret.append(fval)

    return ret

这允许您定义：

def fmeasure_kfold1(array, nfolds):
    ret = []

    # Kfold1 and kfold2 both have this outer loop
    for train_index, test_index in KFold(len(array), nfolds):
        correlation = analyze(array[train_index])

        ret += fmeasure(array[test_index], calc_f,
                        lambda build: get_tests(set(build['modules']), correlation))

    return ret


def fmeasure_kfold2(array, nfolds):
    ret = []

    # Kfold1 and kfold2 both have this outer loop
    for train_index, test_index in KFold(len(array), nfolds):
        correlation = analyze(array[train_index])

        # Retrieved tests is calculated outside the build loop in kfold2
        retrieved_tests = _sum_tests(correlation)

        ret += fmeasure(array[test_index], calc_f, lambda _: retrieved_tests)

    return ret


def fmeasure_all(array):
    return fmeasure(array,
                    lambda relevant, _: calc_f2(relevant),
                    lambda x: x)

到目前为止，fmeasure_kfold1 和 fmeasure_kfold2 看起来非常相似。它们的主要区别在于fmeasure 的调用方式，因此我们可以实现一个通用的fmeasure_kfoldn 函数来集中迭代并收集结果：

def fmeasure_kfoldn(array, nfolds, callable):
    ret = []
    for train_index, test_index in KFold(len(array), nfolds):
        correlation = analyze(array[train_index])
        ret += callable(array[test_index], correlation)
    return ret

这可以很容易地定义fmeasure_kfold1 和fmeasure_kfold2：

def fmeasure_kfold1(array, nfolds):
    def measure(builds, correlation):
        return fmeasure(builds, calc_f, lambda build: get_tests(set(build['modules']), correlation))
    return fmeasure_kfoldn(array, nfolds, measure)


def fmeasure_kfold2(array, nfolds):
    def measure(builds, correlation):
        retrieved_tests = _sum_tests(correlation)
        return fmeasure(builds, calc_f, lambda _: retrieved_tests)
    return fmeasure_kfoldn(array, nfolds, measure)