rc/predavalnice-kontroler
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

functionality for lift control done, will clean it up later

Browse source
This commit is contained in:
0xEmm 2024-03-25 17:55:40 +01:00
parent 7227decc36
commit be85d7ab55
1 changed files with 69 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
projector_motors/projector_motors.py Normal file
View file

@ -0,0 +1,69 @@
import gpiozero.pins.mock
from gpiozero import *
import aiomqtt
import asyncio
# ONLY FOR TESTING ON NON RPI
Device.pin_factory = gpiozero.pins.mock.MockFactory()
relays = [LED(17), LED(27), LED(22), LED(23), LED(5), LED(6), LED(24), LED(25)]
"""
MAIN: SrvDwn SrvUp OpDwn OpUp
0 1 2 3
SIDE: 4 5 6 7
"""
async def task_command2relays(controlClient: aiomqtt.Client):
#relayCtrl = lambda cmd, relay: relays[relay].on() if cmd == 1 else relays[relay].off()
relayCtrl = lambda cmd, relay: [relays[r].on() if cmd == 1 and r == relay else relays[r].off for r in range(len(relays))]
await controlClient.subscribe("p1/projektorji/+/dvigala/#")
async with controlClient.messages() as msgs:
async for mesg in msgs:
mesg: aiomqtt.Message
if mesg.topic.matches('p1/projektorji/+/dvigala/#'):
msgTopicArr = mesg.topic.value.split('/')
cmnd = mesg.payload.decode()
print("Received: [" + str(msgTopicArr) + "] payload: [" + cmnd + "]")
testCase = (msgTopicArr[2], msgTopicArr[4])
match testCase:
case ('glavni', 'servis_dol'):
relayCtrl(cmnd, 0)
print('GLAVNI: SERVIS_DOL: ' + cmnd)
case ('glavni', 'servis_gor'):
relayCtrl(cmnd, 1)
print('GLAVNI: SERVIS_GOR: ' + cmnd)
case ('glavni', 'delovni_dol'):
relayCtrl(cmnd, 2)
print('GLAVNI: DELOVNI_DOL: ' + cmnd)
case ('glavni', 'delovni_gor'):
relayCtrl(cmnd, 3)
print('GLAVNI: DELOVNI_GOR: ' + cmnd)
case ('stranski', 'servis_dol'):
relayCtrl(cmnd, 4)
print('GLAVNI: SERVIS_DOL: ' + cmnd)
case ('stranski', 'servis_gor'):
relayCtrl(cmnd, 5)
print('GLAVNI: SERVIS_GOR: ' + cmnd)
case ('stranski', 'delovni_dol'):
relayCtrl(cmnd, 6)
print('GLAVNI: DELOVNI_DOL: ' + cmnd)
case ('stranski', 'delovni_gor'):
relayCtrl(cmnd, 7)
print('GLAVNI: DELOVNI_GOR: ' + cmnd)
case _:
print("Unrecognized command.")
continue
print("Pushing \'off\' to other relays to prevent conflicts")
await asyncio.sleep(0.01)
async def main():
async with aiomqtt.Client('localhost', 1883) as client:
task_control = asyncio.create_task(task_command2relays(client))
await asyncio.gather(task_control)
if __name__ == '__main__':
asyncio.run(main())