Matthew Frost
e0a0dedea9
All checks were successful
Deploy powerbar.ti Frontend Portal / Deploy-Tinance2-Frontend-Portal-Production (push) Successful in 7s
74 lines
2 KiB
Python
74 lines
2 KiB
Python
from flask_apscheduler import APScheduler
|
|
from app.settings import powerbars
|
|
from app.utils import run_telnet_command
|
|
import time
|
|
|
|
scheduler = APScheduler()
|
|
|
|
def generate_batch_jobs():
|
|
for powerbar in powerbars:
|
|
if powerbars[powerbar]['method'] == 'batch_telnet':
|
|
if powerbars[powerbar]['type'] == 'baytech':
|
|
scheduler.add_job(f"baytech-batch-update-job-{powerbar}", baytech_batch_update, args=[powerbar], trigger='interval', seconds=1, max_instances=1)
|
|
|
|
def baytech_batch_update(powerbar):
|
|
print(f"Running Sync Job: {powerbar}")
|
|
|
|
on_command = "On "
|
|
off_command = "Off "
|
|
|
|
print(f"Checking powerbar state {powerbar}")
|
|
outlets = powerbars[powerbar]['outlets']
|
|
|
|
on_list = []
|
|
off_list = []
|
|
|
|
print (outlets)
|
|
for outlet in outlets:
|
|
|
|
outlet_id = outlet
|
|
outlet_status = powerbars[powerbar]['outlets'][outlet].get('state', 'unknown')
|
|
|
|
if outlet_status == 'on':
|
|
on_list.append(outlet)
|
|
|
|
if outlet_status == 'off':
|
|
off_list.append(outlet)
|
|
|
|
print(f"Creating command sets for: {powerbar}")
|
|
print(f"on_list: {on_list}")
|
|
print(f"off_list: {off_list}")
|
|
|
|
n = 5 # Size of each sublist
|
|
|
|
# Using list comprehension and slicing
|
|
sublist_on = [on_list[i:i + n] for i in range(0, len(on_list), n)]
|
|
sublist_off = [off_list[i:i + n] for i in range(0, len(off_list), n)]
|
|
print(f"sublist_on: {sublist_on}")
|
|
print(f"sublist_off: {sublist_off}")
|
|
|
|
commands = []
|
|
|
|
for list in sublist_on:
|
|
on_command = "On "
|
|
for outlet in list:
|
|
on_command += f"{outlet},"
|
|
commands.append(on_command.rstrip(','))
|
|
|
|
for list in sublist_off:
|
|
off_command = "Off "
|
|
for outlet in list:
|
|
off_command += f"{outlet},"
|
|
commands.append(off_command.rstrip(','))
|
|
|
|
print(f"commands: {commands} for {powerbar}")
|
|
|
|
|
|
|
|
for command in commands:
|
|
print(f"Running command: {command} on {powerbar}")
|
|
run_telnet_command(powerbar, command)
|
|
time.sleep(0.5)
|
|
|
|
print(f"Sync Job Done {powerbar}")
|
|
return
|