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 thepool
to manage and theinterval
elapsed since the last modification. It should either return the newdemand
, orNone
to indicate no change; the latter can also mean that the function does not hit areturn
statement.- \ rule(pool: Pool, interval: float) -> Optional[float]
Note that a rule should not modify the
pool
directly.
- 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
- Parameters:
base – base rule that has no lower bound
rules – lower bound and its control rule
- class cobald.controller.stepwise.Stepwise(*args, **kwargs)[source]
Bases:
Controller
Controller that selects from several strategies based on supply
- See:
UnboundStepwise
allows creatingStepwise
instances via decorators.
- class cobald.controller.stepwise.UnboundStepwise(base: Callable[[Pool, float], float | None])[source]
Bases:
object
Decorator interface for constructing a
Stepwise
controllerApply 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 specificsupply
thresholds usingadd()
.The skeleton can be used like a regular
Controller
: calling it with aPool
and updateinterval
creates aController
instance with the given rules for thePool
.# 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] [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
thresholdRegistration 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
orclass
definitions, and the later forlambda
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 )
- cobald.controller.stepwise.stepwise
alias of
UnboundStepwise