better telnet wrapper
All checks were successful
Deploy powerbar.ti Frontend Portal / Deploy-Tinance2-Frontend-Portal-Production (push) Successful in 7s
All checks were successful
Deploy powerbar.ti Frontend Portal / Deploy-Tinance2-Frontend-Portal-Production (push) Successful in 7s
This commit is contained in:
parent
ea551687f8
commit
34bcb35e6b
2 changed files with 33 additions and 13 deletions
File diff suppressed because one or more lines are too long
44
app/views.py
44
app/views.py
|
@ -1,4 +1,3 @@
|
|||
|
||||
from flask import render_template, Blueprint
|
||||
from app.settings import powerbars, groups
|
||||
from flask import jsonify
|
||||
|
@ -8,6 +7,30 @@ import time
|
|||
|
||||
routes = Blueprint('routes', __name__)
|
||||
|
||||
# Create a dictionary to store the active telnet connections
|
||||
active_telnet_connections = {}
|
||||
|
||||
def get_telnet_connection(powerbar):
|
||||
"""
|
||||
Get the active telnet connection for the given powerbar.
|
||||
If the connection does not exist, create a new one.
|
||||
|
||||
Args:
|
||||
powerbar (str): The name of the powerbar.
|
||||
|
||||
Returns:
|
||||
telnetlib.Telnet: The telnet connection object.
|
||||
|
||||
"""
|
||||
if powerbar in active_telnet_connections:
|
||||
return active_telnet_connections[powerbar]
|
||||
else:
|
||||
tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port'])
|
||||
active_telnet_connections[powerbar] = tn
|
||||
return tn
|
||||
|
||||
|
||||
|
||||
def find_group_by_name(name):
|
||||
for group in groups:
|
||||
if group["name"] == name:
|
||||
|
@ -74,9 +97,8 @@ def powerbar_control(powerbar, outlet, action):
|
|||
if action in ['on', 'off']:
|
||||
try:
|
||||
print(f"Turning {action} powerbar {powerbar} outlet {outlet}")
|
||||
tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port'])
|
||||
tn = get_telnet_connection(powerbar)
|
||||
tn.write(f"{action} {outlet}\r\n".encode('ascii'))
|
||||
tn.close()
|
||||
powerbars[powerbar]['outlets'][outlet]['state'] = action
|
||||
print(f"Turned {action} powerbar {powerbar} outlet {outlet}")
|
||||
except Exception as E:
|
||||
|
@ -84,30 +106,27 @@ def powerbar_control(powerbar, outlet, action):
|
|||
return jsonify({'error': 'Telnet error'}), 500
|
||||
|
||||
if action == 'cycle':
|
||||
tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port'])
|
||||
tn = get_telnet_connection(powerbar)
|
||||
print(f"Turning Off powerbar {powerbar} outlet {outlet}")
|
||||
powerbars[powerbar]['outlets'][outlet]['state'] = "off"
|
||||
tn.write(f"off {outlet}\r\n".encode('ascii'))
|
||||
time.sleep(5)
|
||||
print(f"Turning On powerbar {powerbar} outlet {outlet}")
|
||||
tn.write(f"on {outlet}\r\n".encode('ascii'))
|
||||
tn.close()
|
||||
powerbars[powerbar]['outlets'][outlet]['state'] = "on"
|
||||
|
||||
if action == 'toggle':
|
||||
print(powerbars[powerbar]['outlets'][outlet]['state'])
|
||||
|
||||
if powerbars[powerbar]['outlets'][outlet]['state'] == "on":
|
||||
tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port'])
|
||||
tn = get_telnet_connection(powerbar)
|
||||
print(f"Turning Off powerbar {powerbar} outlet {outlet}")
|
||||
tn.write(f"off {outlet}\r\n".encode('ascii'))
|
||||
tn.close()
|
||||
powerbars[powerbar]['outlets'][outlet]['state'] = "off"
|
||||
elif powerbars[powerbar]['outlets'][outlet]['state'] == "off":
|
||||
tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port'])
|
||||
tn = get_telnet_connection(powerbar)
|
||||
print(f"Turning On powerbar {powerbar} outlet {outlet}")
|
||||
tn.write(f"on {outlet}\r\n".encode('ascii'))
|
||||
tn.close()
|
||||
powerbars[powerbar]['outlets'][outlet]['state'] = "on"
|
||||
|
||||
return jsonify({'state': powerbars[powerbar]['outlets'][outlet].get('state', 'unknown')})
|
||||
|
@ -168,12 +187,11 @@ def powebar_group_action(group, action):
|
|||
try:
|
||||
for device in group["devices"]:
|
||||
powerbar = device["powerbar"]
|
||||
tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port'])
|
||||
tn = get_telnet_connection(powerbar)
|
||||
print(f"Powerbar: {powerbar}")
|
||||
outlets = ",".join([str(x) for x in device['outlets']])
|
||||
print(f"Turning {action} powerbar {powerbar} outlet {outlets}")
|
||||
tn.write(f"{action} {outlets}\r\n".encode('ascii'))
|
||||
tn.close()
|
||||
for outlet in device['outlets']:
|
||||
if outlet in powerbars[powerbar]['outlets']:
|
||||
powerbars[powerbar]['outlets'][outlet]['state'] = action
|
||||
|
@ -194,4 +212,6 @@ def powebar_groups():
|
|||
return jsonify(groups)
|
||||
except Exception as E:
|
||||
print(f"Telnet error: {E}")
|
||||
return jsonify({'error': 'no groups defined for powerbar'}), 404
|
||||
return jsonify({'error': 'no groups defined for powerbar'}), 404
|
||||
|
||||
|
||||
|
|
Loading…
Reference in a new issue