riviera/listeningdaemon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: riviera:d53b51fcd7f77d19dab48a1375a884fa2e3d3083
Branches Tags
riviera:main
..
compare: riviera:bde3a6c6ebf1162098cd29553f0136520e04737e
Branches Tags
riviera:main

No commits in common. "d53b51fcd7f77d19dab48a1375a884fa2e3d3083" and "bde3a6c6ebf1162098cd29553f0136520e04737e" have entirely different histories.
d53b51fcd7 ... bde3a6c6eb

3 changed files with 80 additions and 168 deletions
Show Stats Expand all files Collapse all files

BIN
rule-listener/__pycache__/script.cpython-312.pyc
View File

Binary file not shown.

134
rule-listener/data.py
View File

@ -1,103 +1,47 @@
topics = [
"main/fan/one",
"main/fan/two",
"main/printer/one",
"main/printer/two",
"main/lamp/one",
"main/radio/one"]
walkDelay = 1
baseRule = {
"state": 1,
"count": 1,
"doMessage": 1,
"delay": 30,
"reset": False,
"doTopic": ""
}
rules = {
"main/printer/one": [
{
"description": "Every time a printer is turned on, plan the gesture for 5 minutes later",
"count": 0, # the amount of times
"state": 1, # it has had this state
"doTopic": "main/gesture/printer/one", # i will send to topic
"doMessage": 1, # this message
"delay": 60 * 5, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
},
],
"main/printer/two": [],
"main/lamp/one": [
{
"description": "after the lamp has been on 2 times, another fan should be turned on",
"count": 3, # the amount of times
"state": 1, # it has had this state
"doTopic": "main/fan/two", # i will send to topic
"doMessage": 1, # this message
"delay": False, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
},
],
"main/fan/one": [
{
"description": "after the fan has been on 2 times, another fan should be turned on",
"count": 2, # the amount of times
"state": 1, # it has had this state
"doTopic": "main/fan/two", # i will send to topic
"doMessage": 1, # this message
"delay": False, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
{ # after the fan has been on 2 times, another fan should be turned on
"count": 2, # the amount of times
"state": "on", # it has had this state
"doTopic": "main/fan/two", # i will send to topic
"doMessage": "on", # this message
"delay": False, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
},
# { # After the fan is turned off, the next fan should be turned on
# "count": 4, # the amount of times
# "state": 0, # it has had this state
# "doTopic": "main/fan/two", # i will send to topic
# "doMessage": 1, # this message
# "delay": 4, # after this amount of seconds
# "reset": True, # and i should reset my count afterwards
# },
# { # Turn the fan off every time after 1 minute
# "count": 0,
# "state": 1, # it has had this state
# "doTopic": "main/fan/three", # i will send to topic
# "doMessage": 0, # this message
# "delay": 4, # after this amount of seconds
# "reset": False, # and i should reset my count afterwards
# },
# { # Every 5 minutes, do gesture X
# "state": 1,
# "count": 1,
# "doTopic": "gesture/fan/one",
# "doMessage": 1,
# "delay": 5 * 60,
# "reset": True
# },
],
"main/fan/two": [
{
"description": "after the fan has been on 2 times, another fan should be turned on",
"count": 2, # the amount of times
"state": 1, # it has had this state
"doTopic": "main/fan/one", # i will send to topic
"doMessage": 1, # this message
"delay": False, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
{ # After the fan is turned off, the next fan should be turned on
"count": 4, # the amount of times
"state": "off", # it has had this state
"doTopic": "main/fan/two", # i will send to topic
"doMessage": "on", # this message
"delay": 4, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
},
],
"main/radio/one": [
{
"description": "after the radio has been on 2 times, another fan should be turned on",
"count": 2, # the amount of times
"state": 1, # it has had this state
"doTopic": "main/fan/two", # i will send to topic
"doMessage": 1, # this message
"delay": 10, # after this amount of seconds
"reset": True, # and i should reset my count afterwards
{ # Turn the fan off every time after 1 minute
"count": False, # the amount of times. If its false, just do it.
"state": "on", # it has had this state
"doTopic": "main/fan/three", # i will send to topic
"doMessage": "off", # this message
"delay": 4, # after this amount of seconds
"reset": False, # and i should reset my count afterwards
},
],
{ # Turn the fan on every time after 6 minute
"count": 3, # the amount of times. If its false, just do it.
"state": "on", # it has had this state
"doTopic": "main/fan/three", # i will send to topic
"doMessage": "off", # this message
"delay": 360, # after this amount of seconds
"reset": False, # and i should reset my count afterwards
},
{ # Every 5 minutes, do gesture X
"count": 1,
"state": "on",
"doTopic": "gesture/fan/one",
"doMessage": "on",
"delay": 5 * 60,
"reset": True
}
]
}
@ -109,4 +53,4 @@ gestures = {
"multiplier": 0.7,
"range": [4, 8]
}
}
}

114
rule-listener/script.py
View File

@ -2,138 +2,108 @@ import paho.mqtt.client as mqtt
import asyncio
import secrets
import data
import random
import pprint
from random import randrange
state = {}
rules = data.rules
mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
def publish_transform(fromTopic, toTopic):
if fromTopic and toTopic:
ft = fromTopic.removeprefix("main")
tt = toTopic.removeprefix("main")
path = "transform" + ft + tt
publish(path, 1)
async def do_a_gesture(topic):
print(f"called to do a gesture for {topic}")
# TODO: make sure the data.topics array is shuffled ánd the send in topic is always first
tl = len(data.topics)
theseTopics = data.topics[random.randint(0, tl / 2):random.randint(tl/2, tl)]
wIndex = random.randint(1, tl)
isW = True
print(theseTopics)
while isW:
wIndex+=1
index = wIndex%len(theseTopics)
publish(theseTopics[index], 1)
publish_transform(theseTopics[index - 1], theseTopics[index])
await asyncio.sleep(data.walkDelay)
if index > 0:
publish(theseTopics[index - 1], 0)
print(f"Go off: {theseTopics[index - 1]}")
await asyncio.sleep(data.walkDelay * 0.8)
if(random.random() > 0.8):
# TODO send message that gesture is finished
print("finished the walk")
publish("main/gesture" + topic, 0)
isW = False
async def do_a_gesture(gesture, topic):
state = False
delay = gesture["delay"]
for i in range(randrange(gesture["range"][0],gesture["range"][1])):
await asyncio.sleep(delay)
delay = delay * gesture["multiplier"]
if state:
publish("gesture" + topic, gesture["on"])
state = False
else:
publish("gesture" + topic, gesture["off"])
state = True
publish("main/gesture" + topic, 0)
def on_connect(client, userdata, flags, reason_code, properties):
print(f"Connected with result code {reason_code}")
client.subscribe("main/#")
async def publish_later(rule, transformTopic):
print(
f"publish_later: Processing message with doTopic {rule['doTopic']} after {rule['delay']} seconds delay")
print(f"publish_later: Processing message with doTopic {rule['doTopic']} after {rule['delay']} seconds delay")
await asyncio.sleep(rule['delay'])
del rule['doingDelay']
publish(rule['doTopic'], rule['doMessage'])
publish(transformTopic, 1)
def publish(topic, msg):
print(f"publish {topic} {msg}")
result = mqttc.publish(topic, msg)
status = result[0]
if not status == 0:
print(f"Failed to send message to topic {topic}")
def check_against_rules(msg):
if msg.topic in rules:
print(f"Found {len(rules[msg.topic])} rules on {msg.topic}"),
print(f"Found {len(rules[msg.topic])} rules on this topic"),
willReset = False
for rule in rules[msg.topic]:
print(f"Rule: when the count is > {rule['count']}. Current = {state[msg.topic]['count'][msg.payload]} on payload {msg.payload}"),
for rule in rules[msg.topic]:
if (state[msg.topic]['count'][msg.payload] >= rule['count']
or rule['count'] is None) and int(float(msg.payload)) == rule['state']:
transformTopic = f"transform" + \
msg.topic.removeprefix("main") + rule['doTopic'].removeprefix("main")
print(f"---- rule is met----")
print(rule)
if (rule['reset']):
if (state[msg.topic]['count'][msg.payload] >= rule['count'] or rule['count'] == False) and msg.payload == rule['state']:
print(f"transform topic is {transformTopic}")
if(rule['reset']):
willReset = True
if rule['delay']:
if 'doingDelay' not in rule:
if not 'doingDelay' in rule:
rule['doingDelay'] = True
asyncio.run_coroutine_threadsafe(publish_later(rule, transformTopic), loop)
else:
else:
print('not scheduling')
else:
publish(transformTopic, 1)
publish(rule['doTopic'], rule['doMessage'])
if willReset:
state[msg.topic]['count'][msg.payload] = 0
else:
print("... ... i have no rules for this topic")
# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
print("on message")
if msg.retain:
return False
else:
msg.payload = str(msg.payload.decode("utf-8"))
print(f'msg: {msg.payload} topic: {msg.topic}')
if len(msg.payload) > 140:
print("way to long of a message, not gonna store that")
return False
if msg.topic not in state:
state[msg.topic] = {'count': {}}
state[msg.topic]['count'].setdefault(msg.payload, 0)
state[msg.topic]['count'][msg.payload] += 1
# check if the message has a gesture
if "gesture" in msg.topic:
if msg.payload.strip() != 0:
asyncio.run_coroutine_threadsafe(do_a_gesture(msg.topic.removeprefix('main/gesture')), loop)
if msg.payload.strip() in data.gestures:
asyncio.run_coroutine_threadsafe(do_a_gesture(data.gestures[msg.payload], msg.topic.removeprefix('main/gesture')), loop)
else:
check_against_rules(msg)
async def main():
global loop
loop = asyncio.get_running_loop() # Store the main event loop
mqttc.on_connect = on_connect
mqttc.on_message = on_message
@ -143,11 +113,9 @@ async def main():
# Start the MQTT loop in a background thread
mqttc.loop_start()
# asyncio.run_coroutine_threadsafe(do_a_gesture(data.topics[0].removeprefix('main')), loop)
# # Keep the asyncio event loop running
# Keep the asyncio event loop running
await asyncio.Event().wait()
# Run the main function
asyncio.run(main())
asyncio.run(main())