hausbrand/axmpp.py
2023-04-09 22:09:35 +02:00

93 lines
3 KiB
Python

import logging
import slixmpp
from slixmpp.exceptions import IqError, IqTimeout
import os
from dotenv import load_dotenv
import RPi.GPIO as GPIO
import time
from threading import Thread
import asyncio
GPIO.setmode(GPIO.BCM)
GPIO.setup(23, GPIO.IN, pull_up_down=GPIO.PUD_UP)
init = 0
class EchoBot(slixmpp.ClientXMPP):
def __init__(self, jid, password):
slixmpp.ClientXMPP.__init__(self, jid, password)
self.add_event_handler("session_start", self.session_start)
self.add_event_handler("message", self.message)
loop = asyncio.get_event_loop()
asyncio.run_coroutine_threadsafe(asyncio.to_thread(self.ringLoop), loop)
def session_start(self, event):
self.send_presence()
self.get_roster()
# Most get_*/set_* methods from plugins use Iq stanzas, which
# can generate IqError and IqTimeout exceptions
#
# try:
# self.get_roster()
# except IqError as err:
# logging.error('There was an error getting the roster')
# logging.error(err.iq['error']['condition'])
# self.disconnect()
# except IqTimeout:
# logging.error('Server is taking too long to respond')
# self.disconnect()
def message(self, msg):
global init
if msg['type'] in ('chat', 'normal'):
print (msg['body'])
if msg['body'] == "open":
msg.reply("Buzzing").send()
if init == 0:
GPIO.setup(17, GPIO.OUT)
init = 1
GPIO.output(17, GPIO.LOW)
time.sleep(2)
GPIO.output(17, GPIO.HIGH)
else:
msg.reply("Not Understood").send()
def ringLoop(self):
print("foo1")
#GPIO.setmode(GPIO.BCM) # GPIO Numbers instead of board numbers
recipient = os.getenv("NOTIFY")
#GPIO.setup(23, GPIO.IN, pull_up_down=GPIO.PUD_UP)
while True:
print ("GPIO OUTPUT: ", GPIO.input(23))
time.sleep(0.1)
if (GPIO.input(23) == 0):
self.send_message(mto=recipient, mbody="Klingelt!", mtype='chat')
#self._loop.call_soon_threadsafe(self.send_message(mto=recipient, mbody="Klingelt!", mtype='chat'))
print ("ringing")
print ("ringing")
print ("ringing")
print ("ringing")
print ("ringing")
if __name__ == '__main__':
# Ideally use optparse or argparse to get JID,
# password, and log level.
load_dotenv()
password = os.getenv("PASS")
jid = os.getenv("JID")
#GPIO.setup(17, GPIO.OUT)
#GPIO.output(17, GPIO.LOW)
logging.basicConfig(level=logging.DEBUG, format='%(levelname)-8s %(message)s')
xmpp = EchoBot(jid, password)
#Thread(name="GPIO_INPUT", target=xmpp.ringLoop).start()
#asyncio.run_coroutine_threadsafe(self.waiting_queue.put((data, use_filters)), self.loop)
xmpp.connect()
xmpp.process(forever=True)