Optapy :通过两个连接类使用toList函数- TypeError :没有匹配的重载

Question

我目前正在处理一个员工名册问题，在这个问题中，一个约束的目标是使用约束流来避免员工计划中的空白。这样做的目的是为每个员工获取分配给他们的时间序列以及他们的可用性，在列表中检索这些信息，然后对返回的列表执行检查。

约束流如下：

def continuous_shifts(constraint_factory: ConstraintFactory, score = HardSoftScore.ONE_HARD):
    return constraint_factory \
        .for_each(timeslot_assignment_class) \
        .join(availability_class, [
            Joiners.equal(
                lambda timeslot_assignment : timeslot_assignment.resource.resource_id,
                lambda availability : availability.resource.resource_id
            )
        ]) \
        .group_by(lambda timeslot_assignment, availability : timeslot_assignment.resource.resource_id,
                lambda timeslot_assignment, availability : availability.resource.resource_id,
                ConstraintCollectors.to_list()) \
        .penalize("holes in schedule",score,lambda timeslot_list : holes_in_list(timeslot_list))

我想要做的是加入基于timeslot_assignment属性的可用度，然后根据资源(即雇员)对它们进行分组，然后在列表中返回这些组，在这些列表中我可以测试惩罚部分的进度差距。我必须在可用性类上使用一个联接，因为它们不是作为一个属性包含在资源中，而是为导航目的单独存储的。

我的主要斗争是按职能在小组中返回一份名单。在显示的情况下，我得到了一个错误：

org.optaplanner.constraint.streams.drools.bi.DroolsAbstractBiConstraintStream.groupBy(proxy.PythonBiFunction，proxy.PythonBiFunction，org.optaplanner.core.api.score.stream.DefaultUniConstraintCollector)，没有找到匹配的重载

后面是一个适合ConstraintCollectors的选项列表。

正如我在其他文章中看到的那样，我理解ConstraintCollectors中使用的函数可能存在过载问题，并且可能必须手动指定类型。我尝试了其他组合，如将lambda函数转换为java BiFunctions，或按函数更改组，如下所示：

.group_by(lambda timeslot_assignment, availability : (timeslot_assignment.resource.resource_id,availability.resource.resource_id),
    ConstraintCollectors.to_list())

根据修改后的类/流基数更改错误消息，但没有改进。我还尝试将to_list函数替换为toList，这不会导致任何更改。

我不知道这个问题是来自于我加入/分组的方式，还是更多的是必须指定某些类型的类型问题。

如果能在这个问题上提供任何帮助，我们将不胜感激。

Answer 1

问题是to_list()实际上是to_list(lambda x: x)，它是一个UniConstraintCollector，因此只能与UniConstraintStream (即仅包含一个元素的约束流)一起使用。您有一个BiConstraintStream，所以to_list()不能工作。可以工作的是to_list(lambda x, y: (x, y))，它是一个BiConstraintCollector (因此将与BiConstraintStream一起工作)。因此，约束流应该如下所示：

def continuous_shifts(constraint_factory: ConstraintFactory, score = HardSoftScore.ONE_HARD):
    return constraint_factory \
        .for_each(TimeslotAssignment) \
        .join(Availability,
             Joiners.equal(
                lambda timeslot_assignment : timeslot_assignment.resource.resource_id,
                lambda availability : availability.resource.resource_id
             )
        ) \
        .group_by(ConstraintCollectors.to_list(lambda timeslot_assignment, availability : (timeslot_assignment.resource.resource_id,availability.resource.resource_id))) \
        .penalize("holes in schedule",score,lambda timeslot_list : holes_in_list(timeslot_list))

注意：optapy不再需要使用get_class；您可以直接使用相关的Python类。此外，group‘s不再需要强制转换，您也不需要将Joiners作为列表传递。(虽然旧代码仍然有效)。

def continuous_shifts(constraint_factory: ConstraintFactory, score = HardSoftScore.ONE_HARD):
    return constraint_factory \
        .for_each(TimeslotAssignment) \
        .join(Availability,
             Joiners.equal(
                lambda timeslot_assignment : timeslot_assignment.resource.resource_id,
                lambda availability : availability.resource.resource_id
             )
        ) \
        .group_by(ConstraintCollectors.to_list(lambda timeslot_assignment, availability : (timeslot_assignment.resource.resource_id,availability.resource.resource_id))) \
        .penalize("holes in schedule",score,lambda timeslot_list : holes_in_list(timeslot_list))

就像Lukas说的，to_list是一个性能杀手。如果您正在寻找奇异的空白，我建议查看OptaPy员工计划的快速启动，特别是“两班之间至少两小时”的约束。

关于发现(全局)连续轮班和轮班中断(即需要根据连续轮班/中断的次数对非线性函数进行惩罚/奖励)；最好使用ConstraintCollector来完成；目前在optaplanner-examples中有一个实验性的轮班，但optapy无法访问。一旦该收集器在optaplanner-core中可用，它也将在optapy中可用。您目前无法创建您自己的自定义ConstraintCollector，但它将在未来版本的optapy中可用。