techinc/ti-powerbar
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

less code
All checks were successful
Deploy powerbar.ti Frontend Portal / Deploy-Tinance2-Frontend-Portal-Production (push) Successful in 6s
Details

Browse source
This commit is contained in:
Matthew Frost 2024-02-09 12:46:07 +01:00
parent 2746ac340b
commit 8dfc2e7c33
1 changed files with 0 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

67
app/jobs.py
View file

@ -5,79 +5,12 @@ import time
scheduler = APScheduler() scheduler = APScheduler()
#@scheduler.task('interval', id='do_periodic_serial_job', seconds=3, max_instances=1)
def periodic_serial_job():
print("Running Sync Job")
for powerbar in powerbars:
on_command = "On "
off_command = "Off "
print(f"Checking powerbar state {powerbar}")
outlets = powerbars[powerbar]['outlets']
on_list = []
off_list = []
counter_on = 0
counter_off = 0
max = 9
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"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 command in commands:
print(f"Running command: {command}")
run_telnet_command(powerbar, command)
time.sleep(0.1)
print("Sync Job Done")
def generate_batch_jobs(): def generate_batch_jobs():
for powerbar in powerbars: for powerbar in powerbars:
if powerbars[powerbar]['method'] == 'batch_telnet': if powerbars[powerbar]['method'] == 'batch_telnet':
if powerbars[powerbar]['type'] == 'baytech': if powerbars[powerbar]['type'] == 'baytech':
scheduler.add_job(f"baytech-batch-update-job-{powerbar}", baytech_batch_update, args=[powerbar], trigger='interval', seconds=3, max_instances=1) scheduler.add_job(f"baytech-batch-update-job-{powerbar}", baytech_batch_update, args=[powerbar], trigger='interval', seconds=3, max_instances=1)
def baytech_batch_update(powerbar): def baytech_batch_update(powerbar):
print(f"Running Sync Job: {powerbar}") print(f"Running Sync Job: {powerbar}")