cobald.controller.stepwise module

cobald.controller.stepwise.ControlRule

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], float | None]

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

Bases: object

Container that stores rules for the range of their supply bounds

Parameters:

• base – base rule that has no lower bound

• rules – lower bound and its control rule

get_rule(supply: float)

class cobald.controller.stepwise.Stepwise(*args, **kwargs)

Bases: Controller

Controller that selects from several strategies based on supply

See:

UnboundStepwise allows creating Stepwise instances via decorators.

async run()

Service entry point

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

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
@stepwise
def control(pool: Pool, interval):
    return 10

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

@control.add(supply=100)
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]

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.

@control.add(supply=10)
def linear(pool, interval):
    if pool.utilisation < 0.75:
        return pool.supply - interval
    elif pool.allocation > 0.95:
        return pool.supply + interval

control.add(
    lambda pool, interval: pool.supply * (
        1.2 if pool.allocation > 0.75 else 0.9
    ),
    supply=100
)

rules: List[Tuple[float, Callable[[Pool, float], float | None]]]

s(*args, **kwargs) → Partial[Stepwise]

Create an unbound prototype of this class, partially applying arguments

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

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

Note:

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

cobald.controller.stepwise.stepwise

alias of UnboundStepwise