Source code for cobald.controller.switch

from typing import Union

import trio

from ..interfaces import Pool, Controller
from ..utility import enforce, InvariantError, pairwise
from ..daemon import service

[docs] @service(flavour=trio) class DemandSwitch(Controller): """ Controller that dispatches to slaved controllers based on demand .. code:: python DemandSwitch(pool, linear_control, 10, supply_control) :param target: the pool on which to regulate demand :param default: controller to use by default :param slaves: pairs of minimum demand to switch and corresponding controller :param interval: interval between adjustments in seconds """ def __init__( self, target: Pool, default: Controller, *slaves: Union[float, Controller], interval=1, ): super().__init__(target) enforce( len(slaves) % 2 == 0, InvariantError("slaves must be paired with required demands"), ) self._default = default self._slaves = tuple(sorted(pairwise(slaves))) enforce( all( (isinstance(demand, (int, float)) and isinstance(slave, Controller)) for demand, slave in self._slaves ), InvariantError("slaves must be paired with required demands (float, int)"), ) enforce( in (None, target) and all( in (None, target) for _, slave in self._slaves), InvariantError("slaves must have None or same target as switch target"), ) = target for _, slave in self._slaves: = target self.interval = interval
[docs] async def run(self): while True: self.regulate_demand(self.interval) await trio.sleep(self.interval)
[docs] def regulate(self, interval): chosen = self._default for demand, slave in self._slaves: if demand <= chosen = slave chosen.regulate(interval)