added recorder command backend

recorder_adapters/__init__.py

@@ -0,0 +1,87 @@
+import telnetlib
+from abc import ABC, abstractmethod
+
+# monkey patching of telnet lib
+original_read_until = telnetlib.Telnet.read_until
+original_write = telnetlib.Telnet.write
+
+
+def new_read_until(self, match, timeout=None):
+    if isinstance(match, str):
+        return original_read_until(self, match.encode("ascii"), timeout)
+    else:
+        return original_read_until(self, match, timeout)
+
+
+def new_write(self, buffer):
+    if isinstance(buffer, str):
+        return original_write(self, buffer.encode("ascii"))
+    else:
+        return original_write(self, buffer)
+
+
+telnetlib.Telnet.read_until = new_read_until
+telnetlib.Telnet.write = new_write
+
+
+def read_line(self, timeout=2):
+    return self.read_until("\n", timeout)
+
+
+telnetlib.Telnet.read_line = read_line
+
+
+def read_until_non_empty_line(self):
+    line = self.read_line()
+    if line is None:
+        return None
+    while len(line.rstrip()) <= 0:
+        line = self.read_line()
+    return line
+
+
+telnetlib.Telnet.read_until_non_empty_line = read_until_non_empty_line
+
+
+def assert_string_in_output(self, string, timeout=2):
+    resp = self.read_until(string, timeout)
+    if resp is None:
+        return False, resp,
+    resp = resp.decode("ascii")
+    if string in resp:
+        return True, resp
+    return False, resp
+
+
+telnetlib.Telnet.assert_string_in_output = assert_string_in_output
+
+
+class TelnetAdapter(ABC):
+    def __init__(self, address):
+        self.address = address
+        self.tn = None
+
+    @abstractmethod
+    def login(self):
+        pass
+
+    def run_cmd(self, cmd, timeout=1, auto_connect=True):
+        if self.tn is None and not auto_connect:
+            raise Exception("Not connected!")
+        elif self.tn is None:
+            self.login()
+        self.tn.write(cmd)
+        out = tn.read_until_non_empty_line()
+        res = out
+        while out is not None and out != "":
+            out = tn.read_until_non_empty_line()
+            print(out)
+            res += out
+        return res
+
+    @staticmethod
+    def get_response_str(tn_response):
+        if isinstance(tn_response, bytes):
+            return str(tn_response.decode("ascii").rstrip())
+        else:
+            return str(tn_response).rstrip()

recorder_adapters/extron_smp.py

@@ -0,0 +1,43 @@
+import getpass
+import sys
+from abc import ABC, abstractmethod
+from backend.recorder_adapters import telnetlib, TelnetAdapter
+
+# HOST = "localhost"
+# HOST = "129.13.51.102"  # Audimax SMP 351
+# HOST = "129.13.51.106"  # Tulla SMP 351
+HOST = "172.22.246.207"  # Test SMP MZ
+
+USER = "admin"
+PW = "123mzsmp"
+
+
+class SMP(TelnetAdapter):
+    def __init__(self, address, admin_password):
+        super().__init__(address)
+        self.admin_pw = admin_password
+
+    def login(self):
+        self.tn = telnetlib.Telnet(HOST)
+        self.tn.read_until("\r\nPassword:")
+        # password = getpass.getpass()
+        password = self.admin_pw
+        self.tn.write(password + "\n\r")
+
+        if not self.tn.assert_string_in_output("Login Administrator")[0]:
+            print("WRONG (admin) password!! Exiting!")
+            self.tn = None
+            raise Exception("Could not login as administrator with given pw!")
+        print("OK, we have admin rights!")
+
+    def get_version(self):
+        self.tn.write("1Q\n")
+
+        # print_tn(read_line(tn))
+        return TelnetAdapter.get_response_str(self.tn.read_until_non_empty_line())
+
+
+smp = SMP(HOST, PW)
+print(smp)
+smp.login()
+print(smp.get_version())

recorder_adapters/extron_smp_testing.py

@@ -0,0 +1,65 @@
+import getpass
+import sys
+from abc import ABC, abstractmethod
+from backend.recorder_adapters import telnetlib
+
+# HOST = "localhost"
+# HOST = "129.13.51.102"  # Audimax SMP 351
+# HOST = "129.13.51.106"  # Tulla SMP 351
+HOST = "172.22.246.207"  # Test SMP MZ
+
+user = "admin"
+pw = "123mzsmp"
+
+
+def print_tn(tn_response):
+    if isinstance(tn_response, bytes):
+        print(tn_response.decode("ascii").rstrip())
+    else:
+        print(str(tn_response).rstrip())
+
+
+def run_cmd(tn, cmd, timeout=1):
+    tn.write(cmd)
+    out = tn.read_until_non_empty_line()
+    res = out
+    while out is not None and out != "":
+        out = tn.read_until_non_empty_line()
+        print(out)
+        res += out
+    return res
+
+
+tn = telnetlib.Telnet(HOST)
+tn.read_until("\r\nPassword:")
+# password = getpass.getpass()
+password = pw
+tn.write(password + "\n\r")
+
+if not tn.assert_string_in_output("Login Administrator")[0]:
+    print("WRONG (admin) password!! Exiting!")
+    exit(1)
+
+print("OK, we have admin rights!")
+
+tn.write("1Q\n")
+
+# print_tn(read_line(tn))
+print_tn(tn.read_until_non_empty_line())
+print("test")
+
+
+# print(run_cmd(tn, "I\n"))
+# run_cmd(tn, "X1CERT\n")
+# tn.write(chr(27)+"X1CERT")
+# tn.write("WX1CERT\n")
+# tn.write(chr(27)+"1BOOT")
+# tn.write("W1BOOT\n")
+# print_tn(tn.read_until_non_empty_line())
+# tn.write("I\n")
+# print_tn(tn.read_some())
+
+# tn.write("exit\n\r".encode("ascii"))
+
+# print_tn(tn.read_eager())
+