python - How to implement Timefold constraint with helper function? print() unexpected behavior - Stack Overflow

admin管理员组

文章数量:1279120

The problem: Scheduling workers for two-hour shifts. The objective is to maximize the number of shifts each worker completes within an 8-hour working period. The working period for each employee starts with their first shift and must be followed by a mandatory 10-hour rest period. Additionally, there must be no overlapping shifts assigned to the same worker.

I have based my implementation on the employee_scheduling quickstart project with Python. The domain and core constraints are the same, including:

required_skill(constraint_factory)     
no_overlapping_shifts(constraint_factory)     
unavailable_employee(constraint_factory)     
undesired_day_for_employee(constraint_factory)     
desired_day_for_employee(constraint_factory)

Additionally, I implemented a custom constraint:

enforce_10_hour_rest_period(constraint_factory)

The constraint code:

def enforce_10_hour_rest_period(constraint_factory):
    
    return (
        constraint_factory.for_each(Shift)
        .filter(lambda shift: shift.employee is not None)
        .group_by(
            lambda shift: shift.employee, ConstraintCollectors.to_list())
        .map(lambda employee_name, shifts: calculate_insufficient_breaks(employee_name, sorted(shifts, key=lambda sh: sh.start)))
        .penalize(
            HardSoftDecimalScore.ONE_HARD,
            lambda insufficient_breaks: insufficient_breaks
        )
        .as_constraint("10-hour rest period after 8-hour working period")
    )

The helper function code:

def rest_period_sufficient(last_shift_end, next_shift_start):
    return (next_shift_start - last_shift_end).total_seconds() >= 10 * 3600



def calculate_insufficient_breaks(employee, shifts):
    filtered_shifts = [shift for shift in shifts if shift.employee.name == employee.name]
    l = len(filtered_shifts)
    if len(filtered_shifts) < 2:
        return 0

    insufficient_breaks = 0
    working_period_end = filtered_shifts[0].start + timedelta(hours=8)

    #delete print line
    print(f"working period start: {filtered_shifts[0].start}, working period end {working_period_end}")

    for i in range(1, l):
        #delete print line
        print(f"start: {filtered_shifts[i].start}, end: {filtered_shifts[i].end} ")

        if filtered_shifts[i].end >= working_period_end:
            if not rest_period_sufficient(working_period_end, filtered_shifts[i].start):
            insufficient_breaks += 1
            working_period_end = filtered_shifts[i].start + timedelta(hours=8)
    return insufficient_breaks

Questions:

1. Issue: In the calculate_insufficient_breaks function if I print shift.employee.name values I would receive different employee names. While filtering seems necessary, it is unclear why this occurs given the groupby used in the enforce_10_hour_rest_period function. Furthermore, I observed that removing the first print line - The algorithm (constraint) does not work, the gaps are not 10 hours. And if I remove the second print statement I get the following exception: typeerror: rest_period_sufficient() got an unexpected keyword argument 'hours'.
   The algorithm runs as expected when the print statement remains, but fails otherwise. Why this occurs? Is my current implementation appropriate, or is there any a more efficient approach to defining this constraint?

2. Maximizing number of Shifts Within 8-Hour Windows:
   I aim to maximize the number of shifts a worker completes within their 8-hour work period. My plan was to create a helper function similar to the 10-hour rest period constraint and apply a reward when the maximum number of shifts is achieved within the 8-hour window.
   Is there an optimal way to construct this constraint in Timefold?

The problem: Scheduling workers for two-hour shifts. The objective is to maximize the number of shifts each worker completes within an 8-hour working period. The working period for each employee starts with their first shift and must be followed by a mandatory 10-hour rest period. Additionally, there must be no overlapping shifts assigned to the same worker.

I have based my implementation on the employee_scheduling quickstart project with Python. The domain and core constraints are the same, including:

required_skill(constraint_factory)     
no_overlapping_shifts(constraint_factory)     
unavailable_employee(constraint_factory)     
undesired_day_for_employee(constraint_factory)     
desired_day_for_employee(constraint_factory)

Additionally, I implemented a custom constraint:

enforce_10_hour_rest_period(constraint_factory)

The constraint code:

def enforce_10_hour_rest_period(constraint_factory):
    
    return (
        constraint_factory.for_each(Shift)
        .filter(lambda shift: shift.employee is not None)
        .group_by(
            lambda shift: shift.employee, ConstraintCollectors.to_list())
        .map(lambda employee_name, shifts: calculate_insufficient_breaks(employee_name, sorted(shifts, key=lambda sh: sh.start)))
        .penalize(
            HardSoftDecimalScore.ONE_HARD,
            lambda insufficient_breaks: insufficient_breaks
        )
        .as_constraint("10-hour rest period after 8-hour working period")
    )

The helper function code:

def rest_period_sufficient(last_shift_end, next_shift_start):
    return (next_shift_start - last_shift_end).total_seconds() >= 10 * 3600



def calculate_insufficient_breaks(employee, shifts):
    filtered_shifts = [shift for shift in shifts if shift.employee.name == employee.name]
    l = len(filtered_shifts)
    if len(filtered_shifts) < 2:
        return 0

    insufficient_breaks = 0
    working_period_end = filtered_shifts[0].start + timedelta(hours=8)

    #delete print line
    print(f"working period start: {filtered_shifts[0].start}, working period end {working_period_end}")

    for i in range(1, l):
        #delete print line
        print(f"start: {filtered_shifts[i].start}, end: {filtered_shifts[i].end} ")

        if filtered_shifts[i].end >= working_period_end:
            if not rest_period_sufficient(working_period_end, filtered_shifts[i].start):
            insufficient_breaks += 1
            working_period_end = filtered_shifts[i].start + timedelta(hours=8)
    return insufficient_breaks

Questions:

1. Issue: In the calculate_insufficient_breaks function if I print shift.employee.name values I would receive different employee names. While filtering seems necessary, it is unclear why this occurs given the groupby used in the enforce_10_hour_rest_period function. Furthermore, I observed that removing the first print line - The algorithm (constraint) does not work, the gaps are not 10 hours. And if I remove the second print statement I get the following exception: typeerror: rest_period_sufficient() got an unexpected keyword argument 'hours'.
   The algorithm runs as expected when the print statement remains, but fails otherwise. Why this occurs? Is my current implementation appropriate, or is there any a more efficient approach to defining this constraint?

2. Maximizing number of Shifts Within 8-Hour Windows:
   I aim to maximize the number of shifts a worker completes within their 8-hour work period. My plan was to create a helper function similar to the 10-hour rest period constraint and apply a reward when the maximum number of shifts is achieved within the 8-hour window.
   Is there an optimal way to construct this constraint in Timefold?

• python
• java
• optaplanner
• timefold

Share Improve this question Follow edited Feb 25 at 6:32 Kovy Jacob 1,11988 silver badges2424 bronze badges asked Feb 24 at 18:41 Yorick KramerYorick Kramer 1122 bronze badges

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 1

Both your questions could be solved by using the ConnectedRangeCollector. https://docs.timefold.ai/timefold-solver/latest/constraints-and-score/score-calculation#collectorsConnectedRanges

It allows you to transform events (shifts in your case) to a timeline per employee. The example in the docs shows "equipment" but the idea is the same.

Note how you can transform your Shifts (Jobs in the example) to a grouping of Employee (Equipment in the example) and ConnectedRangeChain. The ConnectedRangeChain then allows you to easily write constraints which check:

• for overlap, no 2 shifts at the same time
• the total length of a ConnectedRange (you want to optimize this to your 8 hours)
• Breaks aka time between shifts (these should be at least 10 hours).

本文标签： pythonHow to implement Timefold constraint with helper function print() unexpected behaviorStack Overflow