cobald.controller.stepwise module


Individual control rule for a pool on a given interval

When a rule for a Stepwise is invoked, it receives the pool to manage and the interval elapsed since the last modification. It should either return the new demand, or None to indicate no change; the latter can also mean that the function does not hit a return statement.

\ rule(pool: Pool, interval: float) -> Optional[float]

Note that a rule should not modify the pool directly.

alias of Callable[[Pool, float], Optional[float]]

class cobald.controller.stepwise.RangeSelector(base: Callable[[Pool, float], float | None], *rules: Tuple[float, Callable[[Pool, float], float | None]])[source]

Bases: object

Container that stores rules for the range of their supply bounds

  • base – base rule that has no lower bound

  • rules – lower bound and its control rule

get_rule(supply: float)[source]
class cobald.controller.stepwise.Stepwise(*args, **kwargs)[source]

Bases: Controller

Controller that selects from several strategies based on supply


UnboundStepwise allows creating Stepwise instances via decorators.

async run()[source]

Service entry point

class cobald.controller.stepwise.UnboundStepwise(base: Callable[[Pool, float], float | None])[source]

Bases: object

Decorator interface for constructing a Stepwise controller

Apply this as a decorator to a ControlRule callable to create a basic controller skeleton. The initial callable forms the base rule. Additional rules can be added for specific supply thresholds using add().

The skeleton can be used like a regular Controller: calling it with a Pool and update interval creates a Controller instance with the given rules for the Pool.

# initial controller skeleton from base case
def control(pool: Pool, interval):
    return 10

# additional rules above specific supply thresholds
def quantized(pool: Pool, interval):
    if pool.utilisation < 0.5:
        return pool.demand - 1
    elif pool.allocation > 0.5:
        return pool.demand + 1

def continuous(pool: Pool, interval):
    if pool.utilisation < 0.5:
        return pool.demand * 1.1
    elif pool.allocation > 0.5:
        return pool.demand * 0.9

# create controller from skeleton
pipeline = control(pool, interval=10)
add(rule: Callable[[Pool, float], float | None], *, supply: float) Callable[[Pool, float], float | None][source]
add(rule: None, *, supply: float) Callable[[Callable[[Pool, float], float | None]], Callable[[Pool, float], float | None]]

Register a new rule above a given supply threshold

Registration supports a single-argument form for use as a decorator, as well as a two-argument form for direct application. Use the former for def or class definitions, and the later for lambda functions and existing callables.

def linear(pool, interval):
    if pool.utilisation < 0.75:
        return - interval
    elif pool.allocation > 0.95:
        return + interval

    lambda pool, interval: * (
        1.2 if pool.allocation > 0.75 else 0.9
s(*args, **kwargs) Partial[Stepwise][source]

Create an unbound prototype of this class, partially applying arguments

def control(pool: Pool, interval):
    return 10

pipeline = control.s(interval=20) >> pool

The partial rules are sealed, and add() cannot be called on it.


alias of UnboundStepwise