Functions¶
Cumulative functions and state functions for resource modeling.
Cumulative Functions¶
CumulFunction Class¶
- class pycsp3_scheduling.functions.cumul_functions.CumulFunction(expressions=<factory>, name=None, _id=-1)[source]¶
Bases:
objectCumulative function representing resource usage over time.
A cumulative function is the sum of elementary cumulative expressions (pulse, step_at_start, step_at_end, step_at). It can be constrained using comparison operators.
- expressions¶
List of elementary cumulative expressions.
- Type:
list[pycsp3_scheduling.functions.cumul_functions.CumulExpr]
Example
>>> tasks = [IntervalVar(size=5, name=f"t{i}") for i in range(3)] >>> demands = [2, 3, 1] >>> usage = CumulFunction() >>> for task, d in zip(tasks, demands): ... usage += pulse(task, d) >>> satisfy(usage <= 5) # Capacity 5
- __init__(expressions=<factory>, name=None, _id=-1)¶
Elementary Functions¶
- pycsp3_scheduling.functions.cumul_functions.pulse(interval, height=None, height_min=None, height_max=None)[source]¶
Create a pulse contribution to a cumulative function.
A pulse represents resource usage during the execution of an interval. The resource is consumed at the specified height from the start to the end of the interval.
- Parameters:
interval (IntervalVar) – The interval variable.
height (int | None) – Fixed height (resource consumption).
height_min (int | None) – Minimum height for variable consumption.
height_max (int | None) – Maximum height for variable consumption.
- Returns:
A CumulExpr representing the pulse.
- Raises:
TypeError – If interval is not an IntervalVar.
ValueError – If height specification is invalid.
- Return type:
CumulExpr
Example
>>> task = IntervalVar(size=10, name="task") >>> p = pulse(task, height=3) # Fixed height 3 >>> p = pulse(task, height_min=1, height_max=5) # Variable height
- pycsp3_scheduling.functions.cumul_functions.step_at(time, height)[source]¶
Create a step contribution at a fixed time point.
The cumulative function increases (or decreases if negative) by the specified height at the given time point and stays at that level.
- Parameters:
- Returns:
A CumulExpr representing the step.
- Raises:
TypeError – If time or height are not integers.
- Return type:
CumulExpr
Example
>>> s = step_at(10, 5) # Increase by 5 at time 10 >>> s = step_at(20, -3) # Decrease by 3 at time 20
- pycsp3_scheduling.functions.cumul_functions.step_at_start(interval, height=None, height_min=None, height_max=None)[source]¶
Create a step contribution at the start of an interval.
The cumulative function increases (or decreases) by the specified height at the start of the interval. The change is permanent.
- Parameters:
interval (IntervalVar) – The interval variable.
height (int | None) – Fixed step height.
height_min (int | None) – Minimum height for variable step.
height_max (int | None) – Maximum height for variable step.
- Returns:
A CumulExpr representing the step at start.
- Raises:
TypeError – If interval is not an IntervalVar.
ValueError – If height specification is invalid.
- Return type:
CumulExpr
Example
>>> task = IntervalVar(size=10, name="task") >>> s = step_at_start(task, height=2) # Increase by 2 at start
- pycsp3_scheduling.functions.cumul_functions.step_at_end(interval, height=None, height_min=None, height_max=None)[source]¶
Create a step contribution at the end of an interval.
The cumulative function increases (or decreases) by the specified height at the end of the interval. The change is permanent.
- Parameters:
interval (IntervalVar) – The interval variable.
height (int | None) – Fixed step height.
height_min (int | None) – Minimum height for variable step.
height_max (int | None) – Maximum height for variable step.
- Returns:
A CumulExpr representing the step at end.
- Raises:
TypeError – If interval is not an IntervalVar.
ValueError – If height specification is invalid.
- Return type:
CumulExpr
Example
>>> task = IntervalVar(size=10, name="task") >>> # Model reservoir: +2 at start (acquire), -2 at end (release) >>> usage = step_at_start(task, 2) + step_at_end(task, -2)
Cumulative Constraints¶
- pycsp3_scheduling.functions.cumul_functions.cumul_range(cumul, min_val, max_val)[source]¶
Constrain a cumulative function to stay within a range.
The cumulative function must satisfy min_val <= cumul <= max_val at all time points.
- Parameters:
cumul (CumulFunction) – The cumulative function.
min_val (int) – Minimum allowed value.
max_val (int) – Maximum allowed value.
- Returns:
A pycsp3-compatible constraint when possible (for simple pulse-based cumulative functions with min_val=0), otherwise a CumulConstraint.
- Raises:
TypeError – If cumul is not a CumulFunction.
ValueError – If min_val > max_val.
Example
>>> tasks = [IntervalVar(size=5, name=f"t{i}") for i in range(3)] >>> usage = sum(pulse(t, 2) for t in tasks) >>> satisfy(cumul_range(usage, 0, 4)) # Between 0 and 4
- pycsp3_scheduling.functions.cumul_functions.always_in(cumul, interval_or_range, min_val, max_val)[source]¶
Constrain cumulative function within a time range.
The cumulative function must satisfy min_val <= cumul <= max_val during the specified interval or fixed time range.
- Parameters:
cumul (CumulFunction) – The cumulative function.
interval_or_range (IntervalVar | tuple[int, int]) – Either an IntervalVar or a (start, end) tuple.
min_val (int) – Minimum allowed value during the range.
max_val (int) – Maximum allowed value during the range.
- Returns:
A CumulConstraint representing the always_in constraint.
- Raises:
TypeError – If arguments have wrong types.
ValueError – If min_val > max_val.
- Return type:
CumulConstraint
Example
>>> usage = sum(pulse(t, 2) for t in tasks) >>> # During maintenance window, only 2 units available >>> satisfy(always_in(usage, (100, 200), 0, 2)) >>> # During task execution, keep minimum level >>> satisfy(always_in(usage, task, 1, 5))
Cumulative Accessors¶
- pycsp3_scheduling.functions.cumul_functions.height_at_start(interval, cumul, absent_value=0)[source]¶
Get the cumulative function height at the start of an interval.
Returns an expression representing the value of the cumulative function at the start time of the interval.
- Parameters:
interval (IntervalVar) – The interval variable.
cumul (CumulFunction) – The cumulative function.
absent_value (int) – Value to use if interval is absent (default: 0).
- Returns:
An expression for the height at interval start.
- Return type:
CumulHeightExpr
Example
>>> usage = sum(pulse(t, 2) for t in tasks) >>> h = height_at_start(task, usage) >>> # h represents the resource level when task starts
- pycsp3_scheduling.functions.cumul_functions.height_at_end(interval, cumul, absent_value=0)[source]¶
Get the cumulative function height at the end of an interval.
Returns an expression representing the value of the cumulative function at the end time of the interval.
- Parameters:
interval (IntervalVar) – The interval variable.
cumul (CumulFunction) – The cumulative function.
absent_value (int) – Value to use if interval is absent (default: 0).
- Returns:
An expression for the height at interval end.
- Return type:
CumulHeightExpr
Example
>>> usage = sum(pulse(t, 2) for t in tasks) >>> h = height_at_end(task, usage) >>> # h represents the resource level when task ends
State Functions¶
StateFunction Class¶
- class pycsp3_scheduling.functions.state_functions.StateFunction(name, transitions=None, initial_state=None, states=None, _id=-1)[source]¶
Bases:
objectState function representing a discrete state over time.
A state function can be in different integer states at different times. Tasks can require specific states during their execution, and transitions between states can have associated times defined by a transition matrix.
- transitions¶
Optional transition matrix defining transition times.
Example
>>> machine = StateFunction(name="machine_mode") >>> # Machine must be in state 2 during task execution >>> satisfy(always_equal(machine, task, 2))
- transitions: TransitionMatrix | None = None¶
- __init__(name, transitions=None, initial_state=None, states=None, _id=-1)¶
TransitionMatrix Class¶
- class pycsp3_scheduling.functions.state_functions.TransitionMatrix(matrix, name=None, _id=-1, FORBIDDEN=-1)[source]¶
Bases:
objectTransition matrix defining valid state transitions and durations.
A transition matrix specifies the time required to transition from one state to another. A value of -1 (or FORBIDDEN) indicates that the transition is not allowed.
- matrix¶
2D list of transition times. matrix[i][j] is the time to transition from state i to state j.
Example
>>> # 3 states with symmetric transition times >>> tm = TransitionMatrix([ ... [0, 5, 10], ... [5, 0, 3], ... [10, 3, 0], ... ]) >>> tm[0, 1] # Time from state 0 to state 1 5
- is_forbidden(from_state, to_state)[source]¶
Check if transition from from_state to to_state is forbidden.
- __init__(matrix, name=None, _id=-1, FORBIDDEN=-1)¶
State Constraints¶
- pycsp3_scheduling.functions.state_functions.always_equal(state_func, interval, value, is_start_aligned=True, is_end_aligned=True)[source]¶
Constrain state function to equal a specific value during interval.
The state function must be equal to the specified value throughout the execution of the interval.
- Parameters:
state_func (StateFunction) – The state function.
interval (IntervalVar) – The interval during which the constraint applies.
value (int) – The required state value.
is_start_aligned (bool) – If True, state must equal value exactly at start.
is_end_aligned (bool) – If True, state must equal value exactly at end.
- Returns:
A StateConstraint representing the always_equal constraint.
- Return type:
StateConstraint
Example
>>> machine = StateFunction(name="machine") >>> # Machine must be in state 2 during task >>> satisfy(always_equal(machine, task, 2))
- pycsp3_scheduling.functions.state_functions.always_in(state_func, interval, min_value, max_value, is_start_aligned=True, is_end_aligned=True)[source]¶
Constrain state function to be within a range during interval.
The state function must be within [min_value, max_value] throughout the execution of the interval.
- Parameters:
state_func (StateFunction) – The state function.
interval (IntervalVar) – The interval during which the constraint applies.
min_value (int) – Minimum allowed state value.
max_value (int) – Maximum allowed state value.
is_start_aligned (bool) – If True, constraint applies exactly at start.
is_end_aligned (bool) – If True, constraint applies exactly at end.
- Returns:
A StateConstraint representing the always_in constraint.
- Return type:
StateConstraint
Example
>>> machine = StateFunction(name="machine") >>> # Machine must be in state 1, 2, or 3 during task >>> satisfy(always_in(machine, task, 1, 3))
- pycsp3_scheduling.functions.state_functions.always_constant(state_func, interval, is_start_aligned=True, is_end_aligned=True)[source]¶
Constrain state function to remain constant during interval.
The state function must not change its value throughout the execution of the interval.
- Parameters:
state_func (StateFunction) – The state function.
interval (IntervalVar) – The interval during which the constraint applies.
is_start_aligned (bool) – If True, constant region starts exactly at start.
is_end_aligned (bool) – If True, constant region ends exactly at end.
- Returns:
A StateConstraint representing the always_constant constraint.
- Return type:
StateConstraint
Example
>>> machine = StateFunction(name="machine") >>> # Machine state cannot change during task >>> satisfy(always_constant(machine, task))
- pycsp3_scheduling.functions.state_functions.always_no_state(state_func, interval, is_start_aligned=True, is_end_aligned=True)[source]¶
Constrain state function to have no defined state during interval.
The state function must not be in any state throughout the execution of the interval (the resource is “unused”).
- Parameters:
state_func (StateFunction) – The state function.
interval (IntervalVar) – The interval during which the constraint applies.
is_start_aligned (bool) – If True, no-state region starts exactly at start.
is_end_aligned (bool) – If True, no-state region ends exactly at end.
- Returns:
A StateConstraint representing the always_no_state constraint.
- Return type:
StateConstraint
Example
>>> machine = StateFunction(name="machine") >>> # Machine must be unused during maintenance >>> satisfy(always_no_state(machine, maintenance_interval))
Function Reference¶
Cumulative Function Types¶
Function |
Description |
|---|---|
|
Rectangular consumption during interval |
|
Permanent step at fixed time |
|
Permanent step at interval start |
|
Permanent step at interval end |
State Constraint Types¶
Constraint |
Description |
|---|---|
|
State equals value during interval |
|
State in range during interval |
|
State doesn’t change during interval |
|
No state defined during interval |
Usage Examples¶
Cumulative Resource Constraint¶
from pycsp3 import satisfy
from pycsp3_scheduling import IntervalVar, pulse
# Tasks with resource demands
tasks = [IntervalVar(size=10, name=f"task{i}") for i in range(5)]
demands = [2, 3, 1, 2, 4]
capacity = 5
# Build cumulative function
resource_usage = sum(pulse(tasks[i], demands[i]) for i in range(5))
# Capacity constraint
satisfy(resource_usage <= capacity)
Variable Height Pulse¶
from pycsp3_scheduling import pulse
# Task with variable resource consumption
task = IntervalVar(size=10, name="task")
p = pulse(task, height_min=1, height_max=5) # Solver chooses height
Reservoir Model (Step Functions)¶
from pycsp3 import satisfy
from pycsp3_scheduling import IntervalVar, step_at_start, step_at_end, cumul_range
tasks = [IntervalVar(size=10, name=f"task{i}") for i in range(3)]
# Each task acquires resource at start, releases at end
reservoir = sum(
step_at_start(task, 1) + step_at_end(task, -1)
for task in tasks
)
# Keep reservoir level between 0 and 2
satisfy(cumul_range(reservoir, 0, 2))
State Function with Transitions¶
from pycsp3 import satisfy
from pycsp3_scheduling import (
IntervalVar, StateFunction, TransitionMatrix, always_equal
)
# Define transition times between states
transitions = TransitionMatrix([
[0, 5, 10], # From state 0: 0→0=0, 0→1=5, 0→2=10
[5, 0, 3], # From state 1
[10, 3, 0], # From state 2
])
machine_mode = StateFunction(name="machine_mode", transitions=transitions)
tasks = [IntervalVar(size=10, name=f"task{i}") for i in range(3)]
required_states = [0, 1, 2]
# Each task requires a specific machine state
for task, state in zip(tasks, required_states):
satisfy(always_equal(machine_mode, task, state))
State Range Constraint¶
from pycsp3_scheduling import StateFunction, always_in
machine = StateFunction(name="machine")
# Task can execute in states 1, 2, or 3
satisfy(always_in(machine, task, min_value=1, max_value=3))
Constant State During Interval¶
from pycsp3_scheduling import StateFunction, always_constant
machine = StateFunction(name="machine")
# Machine state cannot change during task execution
satisfy(always_constant(machine, task))