Matthew Frost
3e9b086c33
All checks were successful
Deploy powerbar.ti Frontend Portal / Deploy-Tinance2-Frontend-Portal-Production (push) Successful in 7s
39 lines
No EOL
1.4 KiB
Python
39 lines
No EOL
1.4 KiB
Python
import threading
|
|
from app.settings import powerbars
|
|
import time
|
|
from app.utils import run_telnet_command
|
|
from queue import Queue
|
|
|
|
queues = {}
|
|
|
|
def baytech_batch_update(powerbar, powerbars, queue):
|
|
while True:
|
|
time.sleep(1)
|
|
print(f"Running Sync Job: {powerbar}")
|
|
print(f"Queue Size: {queue.qsize()}")
|
|
|
|
if queue.qsize() > 0:
|
|
print(f"Queue Size: {queue.qsize()}")
|
|
try:
|
|
job = queue.get()
|
|
print(f"Job: {job}")
|
|
outlet = job['outlet']
|
|
action = job['action']
|
|
print(f"Turning {action} powerbar {powerbar} outlet {outlet}")
|
|
command = run_telnet_command(powerbar,f"{action} {outlet}")
|
|
if command:
|
|
powerbars[powerbar]['outlets'][outlet]['state'] = action
|
|
print(f"Turned {action} powerbar {powerbar} outlet {outlet}")
|
|
except Exception as E:
|
|
print(f"Job Exception Error: {E}")
|
|
pass
|
|
|
|
|
|
def generate_threads():
|
|
for powerbar in powerbars:
|
|
if powerbars[powerbar]['method'] == 'batch_telnet':
|
|
if powerbars[powerbar]['type'] == 'baytech':
|
|
|
|
queues[powerbar] = Queue(maxsize=100)
|
|
t = threading.Thread(target=baytech_batch_update, args=[powerbar, powerbars, queues[powerbar]])
|
|
t.start() |