Home-IoT/craftbeerpi4-pione
1
0
Fork 0
mirror of https://github.com/PiBrewing/craftbeerpi4.git synced 2024-11-10 01:17:42 +01:00
Code Issues Projects Releases Packages Wiki Activity

Add generic and Tasmota specific MQTT actors

Browse source
This commit is contained in:
Philipp Grathwohl 2021-12-03 00:34:08 +01:00
parent 204b02106d
commit f337cb8e97
4 changed files with 112 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

61
cbpi/extension/mqtt_actor/__init__.py
View file

@ -1,61 +1,8 @@
# -*- coding: utf-8 -*-
import asyncio
import json
from cbpi.api import parameters, Property, CBPiActor
from cbpi.api import *
@parameters([Property.Text(label="Topic", configurable=True, description = "MQTT Topic")])
class MQTTActor(CBPiActor):
# Custom property which can be configured by the user
@action("Set Power", parameters=[Property.Number(label="Power", configurable=True,description="Power Setting [0-100]")])
async def setpower(self,Power = 100 ,**kwargs):
self.power=round(Power)
if self.power < 0:
self.power = 0
if self.power > 100:
self.power = 100
await self.set_power(self.power)
def __init__(self, cbpi, id, props):
super(MQTTActor, self).__init__(cbpi, id, props)
async def on_start(self):
self.topic = self.props.get("Topic", None)
self.power = 100
async def on(self, power=None):
if power is not None:
if power != self.power:
power = min(100, power)
power = max(0, power)
self.power = round(power)
await self.cbpi.satellite.publish(self.topic, json.dumps(
{"state": "on", "power": self.power}), True)
self.state = True
pass
async def off(self):
self.state = False
await self.cbpi.satellite.publish(self.topic, json.dumps(
{"state": "off", "power": self.power}), True)
pass
async def run(self):
while self.running:
await asyncio.sleep(1)
def get_state(self):
return self.state
async def set_power(self, power):
self.power = round(power)
if self.state == True:
await self.on(power)
else:
await self.off()
await self.cbpi.actor.actor_update(self.id,power)
pass
from .mqtt_actor import MQTTActor
from .generic_mqtt_actor import GenericMqttActor
from .tasmota_mqtt_actor import TasmotaMqttActor
def setup(cbpi):
'''
@ -67,3 +14,5 @@ def setup(cbpi):
'''
if str(cbpi.static_config.get("mqtt", False)).lower() == "true":
cbpi.plugin.register("MQTTActor", MQTTActor)
cbpi.plugin.register("MQTT Actor (Generic)", GenericMqttActor)
cbpi.plugin.register("MQTT Actor (Tasmota)", TasmotaMqttActor)

37
cbpi/extension/mqtt_actor/generic_mqtt_actor.py Normal file
View file

@ -0,0 +1,37 @@
from cbpi.api import parameters, Property
from . import MQTTActor
@parameters([
Property.Text(label="Topic", configurable=True, description = "MQTT Topic"),
Property.Text(label="Payload", configurable=True, description = "Payload that is sent as MQTT message. Available placeholders are {switch_onoff}: [on|off], {switch_10}: [1|0], {power}: [0-100].")
])
class GenericMqttActor(MQTTActor):
def __init__(self, cbpi, id, props):
MQTTActor.__init__(self, cbpi, id, props)
self.payload = ""
async def on_start(self):
await MQTTActor.on_start(self)
self.payload = self.props.get("Payload", "{{\"state\": \"{switch_onoff}\", \"power\": {power}}}")
def normalize_power_value(self, power):
if power is not None:
if power != self.power:
power = min(100, power)
power = max(0, power)
self.power = round(power)
async def publish_mqtt_message(self, topic, payload):
self.logger.info("Publish '{payload}' to '{topic}'".format(payload = payload, topic = self.topic))
await self.cbpi.satellite.publish(self.topic, payload, True)
async def on(self, power=None):
self.normalize_power_value(power)
formatted_payload = self.payload.format(switch_onoff = "on", switch_10 = 1, power = self.power)
await self.publish_mqtt_message(self.topic, formatted_payload)
self.state = True
async def off(self):
formatted_payload = self.payload.format(switch_onoff = "off", switch_10 = 0, power = self.power)
await self.publish_mqtt_message(self.topic, formatted_payload)
self.state = False

57
cbpi/extension/mqtt_actor/mqtt_actor.py Normal file
View file

@ -0,0 +1,57 @@
import asyncio
import json
from cbpi.api import parameters, Property, CBPiActor
from cbpi.api import *
@parameters([Property.Text(label="Topic", configurable=True, description = "MQTT Topic")])
class MQTTActor(CBPiActor):
# Custom property which can be configured by the user
@action("Set Power", parameters=[Property.Number(label="Power", configurable=True, description="Power Setting [0-100]")])
async def setpower(self,Power = 100 ,**kwargs):
self.power=round(Power)
if self.power < 0:
self.power = 0
if self.power > 100:
self.power = 100
await self.set_power(self.power)
def __init__(self, cbpi, id, props):
super(MQTTActor, self).__init__(cbpi, id, props)
async def on_start(self):
self.topic = self.props.get("Topic", None)
self.power = 100
async def on(self, power=None):
if power is not None:
if power != self.power:
power = min(100, power)
power = max(0, power)
self.power = round(power)
await self.cbpi.satellite.publish(self.topic, json.dumps(
{"state": "on", "power": self.power}), True)
self.state = True
pass
async def off(self):
self.state = False
await self.cbpi.satellite.publish(self.topic, json.dumps(
{"state": "off", "power": self.power}), True)
pass
async def run(self):
while self.running:
await asyncio.sleep(1)
def get_state(self):
return self.state
async def set_power(self, power):
self.power = round(power)
if self.state == True:
await self.on(power)
else:
await self.off()
await self.cbpi.actor.actor_update(self.id,power)
pass

13
cbpi/extension/mqtt_actor/tasmota_mqtt_actor.py Normal file
View file

@ -0,0 +1,13 @@
from cbpi.api import parameters, Property
from . import GenericMqttActor
@parameters([
Property.Text(label="Topic", configurable=True, description = "MQTT Topic"),
])
class TasmotaMqttActor(GenericMqttActor):
def __init__(self, cbpi, id, props):
GenericMqttActor.__init__(self, cbpi, id, props)
async def on_start(self):
await GenericMqttActor.on_start(self)
self.payload = "{switch_onoff}"