Heusor for Python

Heusor is a constraint solver for planning and scheduling problems: school timetables, staff rosters, vehicle routes. You describe the problem with ordinary Python classes and the rules with constraint streams; Heusor searches for the assignment with the best score.

Your lambdas are read from their source and compiled once, when the problem is built. Nothing in Python runs while the solver searches, so a solve runs at engine speed.

Python 3.11 or newer.

Quickstart

A school timetable: lessons need a timeslot and a room, and a teacher cannot teach two lessons at once.

from dataclasses import dataclass
from datetime import time
from typing import Annotated

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


@problem_fact
@dataclass
class Timeslot:
    day: str
    start: time


@problem_fact
@dataclass
class Room:
    name: str


@planning_entity
@dataclass
class Lesson:
    subject: str
    teacher: str
    timeslot: Annotated[Timeslot | None, PlanningVariable] = None
    room: Annotated[Room | None, PlanningVariable] = None


@planning_solution
@dataclass
class Timetable:
    timeslots: list[Timeslot]
    rooms: list[Room]
    lessons: list[Lesson]
    score: HardSoftScore | None = None


@constraint_provider
def constraints(f: ConstraintFactory):
    return [
        f.for_each_unique_pair(Lesson,
                               Joiners.equal(lambda l: l.timeslot),
                               Joiners.equal(lambda l: l.room))
         .penalize(HardSoftScore.ONE_HARD)
         .as_constraint("Room conflict"),
        f.for_each_unique_pair(Lesson,
                               Joiners.equal(lambda l: l.timeslot),
                               Joiners.equal(lambda l: l.teacher))
         .penalize(HardSoftScore.ONE_HARD)
         .as_constraint("Teacher conflict"),
    ]


problem = Timetable(
    timeslots=[Timeslot("MONDAY", time(8, 30)), Timeslot("MONDAY", time(9, 30))],
    rooms=[Room("A"), Room("B")],
    lessons=[Lesson("Math", "Turing"), Lesson("Physics", "Curie"),
             Lesson("Chemistry", "Curie"), Lesson("Biology", "Darwin")],
)

solved = heusor.solve(problem, constraints, time_limit=2)
print(solved.score)                       # HardSoftScore(hard=0, soft=0)
for lesson in solved.lessons:
    print(lesson.subject, lesson.timeslot.start, lesson.room.name)

solve returns a solved copy; problem is not changed.

The domain

Field types: int, bool, str, Enum, datetime, date, time, timedelta, a @problem_fact class, and list / set of those; and list[E] of @planning_entity objects, such as the tasks a task follows. Use integers, not float: scale a quantity (cents, minutes) and declare it int.

Score types: SimpleScore, HardSoftScore, HardMediumSoftScore. A constraint's unit is one level, scaled if you like: HardSoftScore.ONE_HARD, HardSoftScore.of_soft(1000).

Constraints

A @constraint_provider function returns streams, each ending in .penalize(unit, weight) or .reward(unit, weight) and .as_constraint(name).

Existence checks go before a join. After a three- or four-way join, filter, group_by, penalize and reward take the tuple's entities; a match blames each of them. After a join, flatten_last, group_by, complement and load_balance work as they do on single entities; so do complement and load_balance after flatten_last(…).group_by(…). Lambdas after a flatten_last — its filters, keys, collector values and weight — read the entity as it is, its planning variables too.

Collectors whose result is a list. A to_list group's values are read with len(values), x in values and values[i] (None past the end). In to_consecutive_sequences, two members share a sequence when their indices differ by at most one, so two shifts on one day are one sequence member each. A chain is read with len(chain.sequences) and len(chain.breaks). After flatten_last, a sequence has first_index, last_index, length (last_index − first_index + 1), count (members), is_first and is_last; a break has previous_index (the earlier sequence's last), next_index (the later one's first), length (next_index − previous_index), is_first and is_last. With a date index, the indices are dates and the lengths are days. For example, more than five days in a row:

def too_many_days_in_a_row(f):
    return f.for_each(Shift) \
        .group_by(lambda s: s.employee,
                  ConstraintCollectors.to_consecutive_sequences(lambda s: s.day)) \
        .flatten_last(lambda employee, chain: chain.sequences) \
        .filter(lambda employee, seq: seq.length > 5) \
        .penalize(HardSoftScore.ONE_SOFT, lambda employee, seq: seq.length - 5) \
        .as_constraint("Too many days in a row")

A complement after one of these adds the empty list, so it takes no padding; load_balance balances the groups of a number collector only.

Joiners: equal, less_than, less_than_or_equal, greater_than, greater_than_or_equal, overlapping(start, end) (half-open). Each takes one function for both sides, or one per side.

What a lambda may use

• attributes of its arguments — through a planning variable into a fact (lesson.timeslot.day), through a shadow into another entity (visit.previous.location), and items of a list by position (location.times[other.id]) — another entity's list attributes too, through a shadow (stop.previous.times[stop.node], len(…), in), but not its list variable;
• + - * // %, comparisons (chained too), and, or, not, x if c else y, is None, in a list attribute or a constant tuple;
• abs, min, max, len, sum(x.f for x in entity.list_variable);
• max((x.f for x in entity.items), default=k) and min(…) over a list variable or a list of entities, skipping values that are None — in a DerivedShadow, over the entities it depends on;
• dates and times with Python's own meaning: datetime - datetime is a timedelta, .date(), .weekday(), .hour, .total_seconds(), datetime.combine(d, t), timedelta(minutes=x);
• anything that does not mention an argument — a constant, a closure value, timedelta(minutes=30), an enum member — evaluated once, when building;
• calls to your own one-expression functions.

Anything else fails when the constraints are built, never during a solve, with the file, the line and the reason. Where Python and the engine would disagree, Heusor refuses instead of guessing: / (it gives a float — use //), time - time (Python raises — use datetime.combine), and the truth of a str, an enum or a time (compare explicitly).

Solving

heusor.solve(solution, constraints, *, time_limit=None, step_limit=None,
             seed=None, config=None, cancel=None, on_progress=None,
             weights=None)

• time_limit (seconds, construction included) or step_limit; neither means 10 seconds. step_limit with a seed is reproducible.
• cancel=heusor.Cancel(): call cancel.cancel() from another thread to stop early with the best solution so far. The solve releases the interpreter lock.
• on_progress=fn receives progress reports (kind, step, best, elapsed_ms); if it raises, the solve stops and the exception propagates.
• heusor.config_for(solution, constraints, …) returns the exact solver configuration a solve would use; change it and pass it back as config=.
• weights={"Late": HardSoftScore.of_soft(5)} overrides constraint weights by name, as data — one set of constraints for every customer's settings. The weight's non-zero level replaces the constraint's, and each match's weight is multiplied by it; a zero score (HardSoftScore(0, 0)) disables the constraint. heusor.score(…, weights=…) takes the same.

Explaining a score

explanation = heusor.score(solution, constraints)   # weights=… as for solve
explanation.score                 # the total
explanation.constraints           # {name: score}
explanation.indictments[name]     # [Indictment(entity, matches, score), ...]
explanation.blame(entity)         # what that object's matches cost, in total

An indictment names your own objects. A pair charges both of its members and a group every member, so indictments show who is involved; they do not add up to the score.

Errors

Every error is a heusor.HeusorError with a stable code. Messages may be reworded between versions; codes will not.

Licence

Heusor is proprietary software by Shiftmind Limited, free to use, including commercially; it may not be reverse engineered, modified or redistributed on its own. The full terms are in the package's LICENSE, and the notices for the third-party components it includes are in heusor/THIRD_PARTY_LICENSES.txt. Questions: licensing@heusor.com.

Current limits

• Up to four keys per group_by; collectors count, sum, min, max, count_distinct, average, to_list, to_consecutive_sequences and load_balance. A group_by after a to_list, to_consecutive_sequences or load_balance group-by. Joins of up to four streams; flatten_last after a three- or four-way join is not supported yet. An existence check after a join is not supported yet. A flattened sequence offers its indices, length and count, not its members.
• A penalize or reward weight must not be negative: a match that gets one stops the solve with constraint.negative_weight.
• A join to a problem fact is written as an attribute read: shift.employee.skills, not a join with Employee. A stream starts from a @planning_entity class; for_each, for_each_unique_pair, join and existence checks on a problem-fact class are refused.
• A lambda cannot call a method of your classes (visit.total_demand()); write the expression in the lambda, or call a module-level helper whose body is a single return <expression> — no local names, no if statements, no loops. Not compiled either: dict lookups, date.isocalendar() (group by the week's Monday, d - timedelta(days=d.weekday())), float arithmetic (/, 1.3), any() / all() and comprehensions other than sum(x.f for x in e.list_variable).
• An entity is not a value: compare two entities by an attribute (a.id != b.id), and group or collect by one (task.id). is, is not and != between two entities are refused — the error for != suggests is, which is refused too. Facts compare with == / !=; is / is not only against None.
• No expand, map, distinct, concat or flatten (use flatten_last), and no containing, contained_in, containing_any_of or filtering joiners: flatten the list and filter with in, or precompute a count per pair when the problem is built. A join takes one ordering joiner, and not together with overlapping. A group_by takes at least one key.
• PlanningListVariable takes no allows_unassigned: list elements may stay unassigned without it.
• Lambdas must be defined in a file (their source is read).
• Wheels for Linux x86-64 (glibc 2.17 or newer), macOS on Apple silicon (arm64) and Windows x86-64, for Python 3.11 and newer. Not built yet: Linux on ARM (aarch64) and macOS on Intel.