diff --git a/app/__init__.py b/app/__init__.py index f8889c7..31152c9 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -26,7 +26,7 @@ def create_app(): try: print("Is a baytech powerbar, getting outlet status") if powerbar_info['method'] == "telnet": - outlets = get_baytech_status_outlet_telnet(powerbar_info['host'], powerbar_info['port']) + outlets = get_baytech_status_outlet_telnet(powerbar_name) try: for outlet_id, outlet_info in powerbar_info['outlets'].items(): diff --git a/app/settings.py b/app/settings.py index 2fbd8d5..206de31 100644 --- a/app/settings.py +++ b/app/settings.py @@ -18,7 +18,9 @@ powerbars = { "powerbar-aux-space" : { "name" : "Powerbar Aux Space", "host" : "10.209.10.111", - "port" : 3696, + "username" : "admin", + "password" : "admin", + "port" : 2167, "type" : "baytech", "method" : "telnet", "description" : "Powerbar Controlling Aux Space", @@ -39,7 +41,9 @@ powerbars = { "powerbar-main-space" : { "name" : "Powerbar Main Space", "host" : "10.209.10.111", - "port" : 3697, + "username" : "admin", + "password" : "admin", + "port" : 2168, "type" : "baytech", "method" : "telnet", "description" : "Powerbar Controlling Main Space", diff --git a/app/utils.py b/app/utils.py index 5f551d9..794bfd9 100644 --- a/app/utils.py +++ b/app/utils.py @@ -1,15 +1,47 @@ import telnetlib import re +from app.settings import powerbars, groups -def get_baytech_status_outlet_telnet(hostname,port): +def check_input_error(tn): + tn.write("/\r\n".encode('ascii')) + for _ in range(3): + print(f"Checking for input error attempt ${_}") + output = tn.read_until(b"Input error", timeout=3).decode('ascii') + if "Input error" in output: + return True + return False + + +def get_telnet_connection(powerbar): + tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port']) + return tn + + +def run_telnet_command(powerbar, command): + try: + tn = get_telnet_connection(powerbar) + + if check_input_error(tn): + tn.write(f"{command}\r\n".encode('ascii')) + return True + + return False + except Exception as e: + print(f"Telnet error: {e}") + return False + + +def get_baytech_status_outlet_telnet(powerbar): try: # Create a Telnet object and connect to the host - tn = telnetlib.Telnet(hostname, port) - + tn = telnetlib.Telnet(powerbars[powerbar]['host'], powerbars[powerbar]['port']) # Send the command you want to execute command = "status" - tn.write(command.encode('ascii') + b"\r\n") + if check_input_error(tn): + tn.write(command.encode('ascii') + b"\r\n") + else: + return None # Read the output until you receive the prompt or a timeout occurs output = tn.read_until(b"", timeout=5).decode('ascii') # Close the Telnet connection diff --git a/app/views.py b/app/views.py index f680152..65ea678 100644 --- a/app/views.py +++ b/app/views.py @@ -1,5 +1,6 @@ from flask import render_template, Blueprint, request from app.settings import powerbars, groups +from app.utils import run_telnet_command from flask import jsonify import telnetlib import time @@ -10,42 +11,6 @@ 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 run_telnet_command(powerbar, command): - try: - tn = get_telnet_connection(powerbar) - tn.write(f"{command}\r\n".encode('ascii')) - time.sleep(0.2) - tn.write(f"{command}\r\n".encode('ascii')) - return - except Exception as e: - print(f"Telnet error: {e}") - if powerbar in active_telnet_connections: - active_telnet_connections.pop(powerbar) - tn = get_telnet_connection(powerbar) - tn.write(f"{command}\r\n".encode('ascii')) - time.sleep(0.2) - tn.write(f"{command}\r\n".encode('ascii')) - return def find_group_by_name(name): diff --git a/telnet.py b/telnet.py new file mode 100644 index 0000000..b661b80 --- /dev/null +++ b/telnet.py @@ -0,0 +1,14 @@ +import telnetlib + +def check_input_error(): + tn = telnetlib.Telnet("10.209.10.111", 2168) + tn.write("/\r\n".encode('ascii')) + for _ in range(3): + output = tn.read_until(b"Input error", timeout=3).decode('ascii') + print(output) + if "Input error" in output: + return True + return False + +result = check_input_error() +print(result)