Kettle¶
KettleController¶
-
class
core.controller.kettle_controller.
KettleController
(cbpi)¶ Bases:
core.controller.crud_controller.CRUDController
The main kettle controller
-
_is_logic_running
(kettle_id)¶
-
agitator_off
(id)¶
-
agitator_on
(id)¶
-
get_temp
(id)¶
-
get_traget_temp
(id)¶
-
handle_automtic_event
(id, **kwargs)¶ Method to handle the event ‘kettle/+/automatic’
Parameters: - id – The kettle id
- kwargs –
Returns: None
-
heater_off
(id)¶ Convenience Method to switch the heater of a kettle off
Parameters: id – Returns:
-
heater_on
(id)¶ Convenience Method to switch the heater of a kettle on
Parameters: id – the kettle id Returns: (boolean, string)
-
init
()¶ This method initializes all actors during startup. It creates actor instances
Returns:
-
job_stop
(key, **kwargs) → None¶
-
model
¶ alias of
core.database.model.KettleModel
-
toggle_automtic
(id)¶ Convenience Method to toggle automatic
Parameters: id – kettle id as int Returns: (boolean, string)
-
CBPiKettleLogic¶
-
class
core.api.kettle_logic.
CBPiKettleLogic
(*args, **kwds)¶ Bases:
core.api.extension.CBPiExtension
Base Class for a Kettle logic.
-
init
()¶ Code which will be executed when the logic is initialised. Needs to be overwritten by the implementing logic
Returns: None
-
run
()¶ This method is running as background process when logic is started. Typically a while loop responsible that the method keeps running
- while self.running:
- await asyncio.sleep(1)
Returns: None
-
stop
()¶ Code which will be executed when the logic is stopped. Needs to be overwritten by the implementing logic
Returns: None
-
Custom Logic¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | import asyncio
from core.api import Property, on_event
from core.api.kettle_logic import CBPiKettleLogic
class CustomLogic(CBPiKettleLogic):
test = Property.Number(label="Test")
running = True
async def wait_for_event(self, topic, callback=None, timeout=None):
future_obj = self.cbpi.app.loop.create_future()
async def default_callback(id, **kwargs):
future_obj.set_result("HELLO")
if callback is None:
self.cbpi.bus.register(topic=topic, method=default_callback)
else:
callback.future = future_obj
self.cbpi.bus.register(topic=topic, method=callback)
if timeout is not None:
try:
print("----> WAIT FOR FUTURE")
await asyncio.wait_for(future_obj, timeout=timeout)
print("------> TIMEOUT")
return future_obj.result()
except asyncio.TimeoutError:
print('timeout!')
else:
print("----> WAIT FOR FUTURE")
await future_obj
return future_obj.result()
async def run(self):
async def my_callback(value, **kwargs):
if value == 5:
self.cbpi.bus.unregister(my_callback)
kwargs["future"].set_result("AMAZING")
else:
print("OTHER VALUE", value)
result = await self.wait_for_event("sensor/1", callback=my_callback)
print("THE RESULT", result)
'''
while self.running:
print("RUN", self.test)
value = await self.cbpi.sensor.get_value(1)
print(value)
if value >= 10:
break
await asyncio.sleep(1)
'''
print("YES IM FINISHED STOP LOGIC")
def setup(cbpi):
'''
This method is called by the server during startup
Here you need to register your plugins at the server
:param cbpi: the cbpi core
:return:
'''
cbpi.plugin.register("CustomKettleLogic", CustomLogic)
|
config.yaml
1 2 | name: DummyKettleLogic
version: 4
|