Constraint cookbook (Python)

One recipe per common rule: skills, availability, capacity, overlapping shifts, rest time, fairness and more. Each recipe is a complete function. Find the rule closest to yours, copy it, and adapt it.

If you have not read Getting started, start there: it explains for_each, filter, joiners and penalize. For every stream step and what a lambda may use, see the package reference.

The model the recipes use

Workers, and jobs with a day, a start and an end. Each job needs one worker. Times are minutes from the start of the schedule.

from dataclasses import dataclass, field
from typing import Annotated

from heusor import (
    ConstraintCollectors, ConstraintFactory, HardSoftScore, Joiners,
    PlanningVariable, planning_entity, planning_solution, problem_fact,
)

MIN_REST = 11 * 60  # shortest gap between two jobs of one worker, in minutes
MAX_PER_DAY = 2  # most jobs one worker may do in a day


@problem_fact
@dataclass
class Worker:
    name: str
    skills: set[str] = field(default_factory=set)
    unavailable_days: set[int] = field(default_factory=set)
    max_hours: int = 40


@planning_entity
@dataclass
class Job:
    id: int
    day: int
    start: int  # minutes
    end: int  # minutes
    skill: str
    team: int | None = None  # jobs of one team go to one worker
    preferred: Worker | None = None  # a worker who would like this job
    conflicts: set[int] = field(default_factory=set)  # ids of jobs it must not share a worker with
    worker: Annotated[Worker | None, PlanningVariable] = None


@planning_solution
@dataclass
class Plan:
    workers: list[Worker]
    jobs: list[Job]
    score: HardSoftScore | None = None

A constraint provider returns the recipes you use:

@constraint_provider
def constraints(f: ConstraintFactory):
    return [required_skill(f), unavailable_day(f), capacity(f), no_overlap(f)]

Rules about one job

The simplest shape: stream every assigned job, keep the ones breaking the rule, penalise each.

The worker must have the required skill

def required_skill(f: ConstraintFactory):
    """Hard: the worker has the skill the job needs."""
    return (
        f.for_each(Job)
        .filter(lambda job: job.skill not in job.worker.skills)
        .penalize(HardSoftScore.ONE_HARD)
        .as_constraint("Required skill")
    )

for_each streams only jobs whose worker is assigned. An unassigned job matches no recipe here: construction starts from nothing assigned, and a rule that penalised emptiness would distort it.

Nobody works on a day they are unavailable

def unavailable_day(f: ConstraintFactory):
    """Hard: nobody works on a day they are unavailable."""
    return (
        f.for_each(Job)
        .filter(lambda job: job.day in job.worker.unavailable_days)
        .penalize(HardSoftScore.ONE_HARD)
        .as_constraint("Unavailable day")
    )

Rewarding a preference

reward adds instead of subtracting. Use it for things you want more of.

def preference(f: ConstraintFactory):
    """Soft: reward giving a job to the worker who asked for it."""
    return (
        f.for_each(Job)
        .filter(lambda job: job.preferred is not None and job.worker == job.preferred)
        .reward(HardSoftScore.ONE_SOFT)
        .as_constraint("Preference")
    )

Rules about totals: group, collect, compare

Group the jobs by a key, run a collector over each group, and keep the groups over a limit. The collector is maintained incrementally: moving one job updates two groups, not all of them.

Capacity: total hours within a maximum

def capacity(f: ConstraintFactory):
    """Hard: a worker's hours stay within their maximum, penalised by the
    overshoot."""
    return (
        f.for_each(Job)
        .group_by(lambda job: job.worker,
                  ConstraintCollectors.sum(lambda job: (job.end - job.start) // 60))
        .filter(lambda worker, hours: hours > worker.max_hours)
        .penalize(HardSoftScore.ONE_HARD, lambda worker, hours: hours - worker.max_hours)
        .as_constraint("Capacity")
    )

The group's key is the worker itself, so the filter and the weight can read its fields (worker.max_hours).

Penalise by how much a limit is exceeded, not a flat 1. "One hour over" then scores better than "ten hours over", which gives the search a direction back to feasibility. The same advice applies to every limit below.

At most N per day: group by two keys

Give group_by several keys — here the worker and the day — and it makes one group per combination. The filter and the weight receive the keys in order, then the collected value.

def max_per_day(f: ConstraintFactory):
    """Hard: at most MAX_PER_DAY jobs per worker per day; each job over the
    limit costs one."""
    return (
        f.for_each(Job)
        .group_by(lambda job: job.worker, lambda job: job.day, ConstraintCollectors.count())
        .filter(lambda worker, day, n: n > MAX_PER_DAY)
        .penalize(HardSoftScore.ONE_HARD, lambda worker, day, n: n - MAX_PER_DAY)
        .as_constraint("Max per day")
    )

Up to four keys. ConstraintCollectors.count_distinct(f) counts distinct values instead — for example the number of different days a worker works.

Fairness: spread the load, idle workers included

load_balance scores how uneven the group values are. complement(Worker) first adds a group with count 0 for every worker who has no job, so an idle worker counts as unfair too.

def fairness(f: ConstraintFactory):
    """Soft: spread jobs evenly over every worker, idle ones included."""
    return (
        f.for_each(Job)
        .group_by(lambda job: job.worker, ConstraintCollectors.count())
        .complement(Worker)
        .group_by(ConstraintCollectors.load_balance(scale=1000))
        .penalize(HardSoftScore.ONE_SOFT, lambda balance: balance.unfairness)
        .as_constraint("Fairness")
    )

The unfairness is sqrt(Σ(count − mean)²), times scale and rounded. The scale keeps the result an integer without losing the difference between "slightly" and "very" uneven.

Rules about pairs: join, then compare

A pair rule looks at two jobs at once. for_each_unique_pair pairs jobs that agree on every joiner, each pair once. Partners are found through an index on the joiners, not by comparing every job with every other, so these rules stay fast on large problems. Make the joiners as narrow as the rule allows: all of these join on the worker, because a pair on different workers can never break them.

No overlap in time

def no_overlap(f: ConstraintFactory):
    """Hard: one worker's jobs do not overlap, penalised by the minutes of
    overlap."""
    return (
        f.for_each_unique_pair(
            Job,
            Joiners.equal(lambda job: job.worker),
            Joiners.overlapping(lambda job: job.start, lambda job: job.end),
        )
        .penalize(HardSoftScore.ONE_HARD,
                  lambda a, b: min(a.end, b.end) - max(a.start, b.start))
        .as_constraint("No overlap")
    )

overlapping treats each job as the half-open interval [start, end). Jobs that only touch — one ends at 16:00, the next starts at 16:00 — do not overlap.

Minimum rest between jobs

def minimum_rest(f: ConstraintFactory):
    """Hard: at least MIN_REST minutes between two jobs of one worker,
    penalised by the minutes short. Overlapping jobs are left to no_overlap."""
    return (
        f.for_each(Job)
        .join(
            Job,
            Joiners.equal(lambda job: job.worker),
            Joiners.less_than_or_equal(lambda job: job.end, lambda job: job.start),
        )
        .filter(lambda first, second: second.start - first.end < MIN_REST)
        .penalize(HardSoftScore.ONE_HARD,
                  lambda first, second: MIN_REST - (second.start - first.end))
        .as_constraint("Minimum rest")
    )

join with less_than_or_equal(end, start) pairs each job with the jobs that start after it ends, so each pair is seen once and in order. Keep rules independent: this one never sees overlapping jobs, which no_overlap owns. A solution then breaks each rule for one reason, and the explanation says which.

Keep two jobs apart

def keep_apart(f: ConstraintFactory):
    """Hard: two conflicting jobs do not go to the same worker."""
    return (
        f.for_each_unique_pair(Job, Joiners.equal(lambda job: job.worker))
        .filter(lambda a, b: b.id in a.conflicts or a.id in b.conflicts)
        .penalize(HardSoftScore.ONE_HARD)
        .as_constraint("Keep apart")
    )

Keep a team's jobs together

The joiner here is the team, not the worker: pair each team's jobs, and penalise the pairs given to different workers.

def keep_together(f: ConstraintFactory):
    """Hard: jobs of one team go to the same worker."""
    return (
        f.for_each_unique_pair(Job, Joiners.equal(lambda job: job.team))
        .filter(lambda a, b: a.team is not None and a.worker != b.worker)
        .penalize(HardSoftScore.ONE_HARD)
        .as_constraint("Keep together")
    )

Jobs without a team share the key None, so the filter drops those pairs.

How these recipes are checked

The test that runs this page's code and scores each recipe on small hand-made plans. For example, jobs from 08:00 to 16:00 and from 12:00 to 20:00 on one worker cost 240: four hours of overlap, in minutes.

When you write a rule of your own, score a few hand-made cases with heusor.score(plan, constraints) and compare each rule's share, explanation.constraints[name], with what you expect. It is the quickest way to find a rule that does not say what you meant.

Not covered yet

• Time windows and travel time on routes use list variables and shadow variables: see the vehicle-routing example.
• Joins between two different classes (.join(Other, …)) and existence checks (.if_exists, .if_not_exists) follow the same pattern as the pair rules; see the package reference.