code to call recorder functions from frointend

This commit is contained in:
2019-12-12 18:44:07 +01:00
parent da200f95b8
commit bb4db25dcd
6 changed files with 160 additions and 79 deletions

View File

@@ -58,7 +58,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
# print(self.tn.assert_string_in_output("Login Administrator"))
self.tn = None
logger.error("Could definitely not login (as admin) with given password! {}".format(self.address))
raise Exception("Could not login as administrator with given pw!")
raise LrcException("Could not login as administrator with given pw!")
# print("OK, we have admin rights!")
def _get_name(self):
@@ -158,7 +158,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if mode not in range(4):
raise Exception("Only values from 0 to 3 are allowed!")
raise LrcException("Only values from 0 to 3 are allowed!")
self.tn.write(self.esc_char + str(mode) + "CV\n")
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -175,7 +175,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if mode not in [0, 2]:
raise Exception("Only values 0 and 2 are allowed!")
raise LrcException("Only values 0 and 2 are allowed!")
self.tn.write(self.esc_char + "1*{}XF\n".format(mode))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -188,7 +188,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if mode not in [0, 2]:
raise Exception("Only values 0 and 2 are allowed!")
raise LrcException("Only values 0 and 2 are allowed!")
self.tn.write(self.esc_char + "0*{}XF\n".format(mode))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -242,7 +242,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if mode not in range(4):
raise Exception("Only values from 0 to 3 are allowed!")
raise LrcException("Only values from 0 to 3 are allowed!")
self.tn.write(str(mode) + "X\n")
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -306,15 +306,15 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if input_num not in range(1, 6):
raise Exception("input_num must be a value between 1 and 5!")
raise LrcException("input_num must be a value between 1 and 5!")
if channel_num not in range(1, 3):
raise Exception("input_num must be a value between 1 and 2!")
raise LrcException("input_num must be a value between 1 and 2!")
self.tn.write("{}*{}!\n".format(input_num, channel_num))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_input(self, channel_num: int):
if channel_num not in range(1, 2):
raise Exception("input_num must be a value between 1 and 2!")
raise LrcException("input_num must be a value between 1 and 2!")
self.tn.write("{}!\n".format(channel_num))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -329,33 +329,33 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if input_num not in range(1, 6):
raise Exception("input_num must be a value between 1 and 5!")
raise LrcException("input_num must be a value between 1 and 5!")
if input_format not in range(1, 4):
raise Exception("input_num must be a value between 1 and 3!")
raise LrcException("input_num must be a value between 1 and 3!")
self.tn.write("{}*{}\\\n".format(input_num, input_format))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_input_format(self, input_num: int):
if input_num not in range(1, 6):
raise Exception("input_num must be a value between 1 and 5!")
raise LrcException("input_num must be a value between 1 and 5!")
self.tn.write("{}\\\n".format(input_num))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_input_name(self, input_num: int, input_name: str):
if input_num not in range(1, 6):
raise Exception("input_num must be a value between 1 and 5!")
raise LrcException("input_num must be a value between 1 and 5!")
if len(input_name) > 16:
raise Exception("input_name must be no longer than 16 chars")
raise LrcException("input_name must be no longer than 16 chars")
try:
input_name.encode('ascii')
except UnicodeEncodeError:
raise Exception("input_name must only contain ascii characters")
raise LrcException("input_name must only contain ascii characters")
self.tn.write("{}{},{}NI\n".format(self.esc_char, input_num, input_name))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_input_name(self, input_num: int):
if input_num not in range(1, 6):
raise Exception("input_num must be a value between 1 and 5!")
raise LrcException("input_num must be a value between 1 and 5!")
self.tn.write("{}{}NI\n".format(self.esc_char, input_num))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -401,7 +401,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if extension_time not in range(0, 100):
raise Exception("extension_time must be a value between 0 and 99!")
raise LrcException("extension_time must be a value between 0 and 99!")
self.tn.write("{}E{}RCDR\n".format(self.esc_char, extension_time))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -435,82 +435,82 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
def recall_user_preset(self, channel_number: int, preset_number: int):
if channel_number not in range(1, 3):
raise Exception("channel_number must be a value between 1 and 2!")
raise LrcException("channel_number must be a value between 1 and 2!")
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("1*{}*{}.\n".format(channel_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def save_user_preset(self, channel_number: int, preset_number: int):
if channel_number not in range(1, 3):
raise Exception("channel_number must be a value between 1 and 2!")
raise LrcException("channel_number must be a value between 1 and 2!")
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("1*{}*{},\n".format(channel_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_user_preset_name(self, preset_number: int, preset_name: str):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
if len(preset_name) > 16:
raise Exception("preset_name must be no longer than 16 chars")
raise LrcException("preset_name must be no longer than 16 chars")
try:
preset_name.encode('ascii')
except UnicodeEncodeError:
raise Exception("preset_name must only contain ascii characters")
raise LrcException("preset_name must only contain ascii characters")
self.tn.write("{}1*{},{}PNAM\n".format(self.esc_char, preset_number, preset_name))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_user_preset_name(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}1*{}PNAM\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_user_presets(self, input_number: int):
if input_number not in range(1, 6):
raise Exception("input_number must be a value between 1 and 5!")
raise LrcException("input_number must be a value between 1 and 5!")
self.tn.write("52*{}#\n".format(input_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
# Input Presets
def recall_input_preset(self, channel_number: int, preset_number: int):
if channel_number not in range(1, 3):
raise Exception("channel_number must be a value between 1 and 2!")
raise LrcException("channel_number must be a value between 1 and 2!")
if preset_number not in range(1, 129):
raise Exception("preset_number must be a value between 1 and 128!")
raise LrcException("preset_number must be a value between 1 and 128!")
self.tn.write("2*{}*{}.\n".format(channel_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def save_input_preset(self, channel_number: int, preset_number: int):
if channel_number not in range(1, 3):
raise Exception("channel_number must be a value between 1 and 2!")
raise LrcException("channel_number must be a value between 1 and 2!")
if preset_number not in range(1, 129):
raise Exception("preset_number must be a value between 1 and 128!")
raise LrcException("preset_number must be a value between 1 and 128!")
self.tn.write("1*{}*{},\n".format(channel_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_input_preset_name(self, preset_number: int, preset_name: str):
if preset_number not in range(1, 129):
raise Exception("preset_number must be a value between 1 and 128!")
raise LrcException("preset_number must be a value between 1 and 128!")
if len(preset_name) > 16:
raise Exception("preset_name must be no longer than 16 chars")
raise LrcException("preset_name must be no longer than 16 chars")
try:
preset_name.encode('ascii')
except UnicodeEncodeError:
raise Exception("preset_name must only contain ascii characters")
raise LrcException("preset_name must only contain ascii characters")
self.tn.write("{}2*{},{}PNAM\n".format(self.esc_char, preset_number, preset_name))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_input_preset_name(self, preset_number: int):
if preset_number not in range(1, 129):
raise Exception("preset_number must be a value between 1 and 128!")
raise LrcException("preset_number must be a value between 1 and 128!")
self.tn.write("{}2*{}PNAM\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def delete_input_preset(self, preset_number: int):
if preset_number not in range(1, 129):
raise Exception("preset_number must be a value between 1 and 128!")
raise LrcException("preset_number must be a value between 1 and 128!")
self.tn.write("{}X2*{}PRST\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -530,9 +530,9 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 4):
raise Exception("output_number must be a value between 1 and 3!")
raise LrcException("output_number must be a value between 1 and 3!")
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("3*{}*{}.\n".format(output_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -547,33 +547,33 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 4):
raise Exception("output_number must be a value between 1 and 3!")
raise LrcException("output_number must be a value between 1 and 3!")
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("3*{}*{},\n".format(output_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_streaming_preset_name(self, preset_number: int, preset_name: str):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
if len(preset_name) > 16:
raise Exception("preset_name must be no longer than 16 chars")
raise LrcException("preset_name must be no longer than 16 chars")
try:
preset_name.encode('ascii')
except UnicodeEncodeError:
raise Exception("preset_name must only contain ascii characters")
raise LrcException("preset_name must only contain ascii characters")
self.tn.write("{}3*{},{}PNAM\n".format(self.esc_char, preset_number, preset_name))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_streaming_preset_name(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}3*{}PNAM\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def reset_streaming_preset_to_default(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}X3*{}PRST\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -589,9 +589,9 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 4):
raise Exception("output_number must be a value between 1 and 3!")
raise LrcException("output_number must be a value between 1 and 3!")
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("4*{}*{}.\n".format(output_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -606,46 +606,46 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 4):
raise Exception("output_number must be a value between 1 and 3!")
raise LrcException("output_number must be a value between 1 and 3!")
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("4*{}*{},\n".format(output_number, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_encoder_preset_name(self, preset_number: int, preset_name: str):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
if len(preset_name) > 16:
raise Exception("preset_name must be no longer than 16 chars")
raise LrcException("preset_name must be no longer than 16 chars")
try:
preset_name.encode('ascii')
except UnicodeEncodeError:
raise Exception("preset_name must only contain ascii characters")
raise LrcException("preset_name must only contain ascii characters")
self.tn.write("{}4*{},{}PNAM\n".format(self.esc_char, preset_number, preset_name))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_encoder_preset_name(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}4*{}PNAM\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def reset_encoder_preset_to_default(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}X4*{}PRST\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
# Layout Presets
def save_layout_preset(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("7*{},\n".format(preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def recall_layout_preset(self, preset_number: int, include_input_selections: bool = True):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
if include_input_selections:
self.tn.write("7*{}.\n".format(preset_number))
else:
@@ -654,25 +654,25 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
def set_layout_preset_name(self, preset_number: int, preset_name: str):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
if len(preset_name) > 16:
raise Exception("preset_name must be no longer than 16 chars")
raise LrcException("preset_name must be no longer than 16 chars")
try:
preset_name.encode('ascii')
except UnicodeEncodeError:
raise Exception("preset_name must only contain ascii characters")
raise LrcException("preset_name must only contain ascii characters")
self.tn.write("{}7*{},{}PNAM\n".format(self.esc_char, preset_number, preset_name))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_layout_preset_name(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}7*{}PNAM\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def reset_layout_preset_to_default(self, preset_number: int):
if preset_number not in range(1, 17):
raise Exception("preset_number must be a value between 1 and 16!")
raise LrcException("preset_number must be a value between 1 and 16!")
self.tn.write("{}X7*{}PRST\n".format(self.esc_char, preset_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -690,7 +690,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 3):
raise Exception("output_number must be a value between 1 and 2!")
raise LrcException("output_number must be a value between 1 and 2!")
self.tn.write("{}*1B\n".format(output_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -703,7 +703,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 3):
raise Exception("output_number must be a value between 1 and 2!")
raise LrcException("output_number must be a value between 1 and 2!")
self.tn.write("{}*0B\n".format(output_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
@@ -716,7 +716,7 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if output_number not in range(1, 3):
raise Exception("output_number must be a value between 1 and 2!")
raise LrcException("output_number must be a value between 1 and 2!")
self.tn.write("{}B\n".format(output_number))
return int(TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())) > 0
@@ -741,25 +741,25 @@ class SMP35x(TelnetAdapter, RecorderAdapter):
:return:
"""
if input_number not in range(1, 6):
raise Exception("input_number must be a value between 1 and 6!")
raise LrcException("input_number must be a value between 1 and 6!")
self.tn.write("{}I{}HDCP\n".format(self.esc_char, input_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_input_authorization_hdcp_on(self, input_number: int):
if input_number not in range(1, 6):
raise Exception("input_number must be a value between 1 and 6!")
raise LrcException("input_number must be a value between 1 and 6!")
self.tn.write("{}E{}*1HDCP\n".format(self.esc_char, input_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def set_input_authorization_hdcp_off(self, input_number: int):
if input_number not in range(1, 6):
raise Exception("input_number must be a value between 1 and 6!")
raise LrcException("input_number must be a value between 1 and 6!")
self.tn.write("{}E{}*0HDCP\n".format(self.esc_char, input_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def get_input_authorization_hdcp_status(self, input_number: int):
if input_number not in range(1, 6):
raise Exception("input_number must be a value between 1 and 6!")
raise LrcException("input_number must be a value between 1 and 6!")
self.tn.write("{}E{}HDCP\n".format(self.esc_char, input_number))
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())

View File

@@ -0,0 +1,61 @@
import logging
from typing import Union
from backend.models import Recorder, RecorderCommand, LrcException
from backend.recorder_adapters import get_recorder_adapter_by_id
logger = logging.getLogger("lrc.recorder_adapters.helpers")
def validate_recorder_command_params(recorder_command: RecorderCommand, params: dict, fail_on_missing_params=False):
if recorder_command.parameters is None or len(recorder_command.parameters) == 0:
if len(params) == 0:
logger.debug("Number (0) of parameters matching expected number of args of command signature.")
return True
logger.info("More arguments specified ({}) than expected!".format(len(params)))
return False
for p_n in recorder_command.parameters:
p_t = recorder_command.parameters[p_n]
p = params.get(p_n, None)
if p is None:
if fail_on_missing_params:
return False
else:
if p_t == 'int':
params[p_n] = int(p)
return True
def get_function_from_recorder_command(recorder: Recorder, recorder_command: RecorderCommand,
connect_by_network_name=True):
adapter_name, function_name = recorder_command.name.split(':')
if connect_by_network_name and recorder.network_name is not None:
address = recorder.network_name
elif recorder.ip6 is not None:
address = recorder.ip6
else:
address = recorder.ip
logger.debug("Using {} to create recorder adapter to connect to {}.".format(address, recorder.name))
package_name, class_name = adapter_name.split('.')
adapter = get_recorder_adapter_by_id(class_name, address=address, user=recorder.username,
password=recorder.password, firmware_version=recorder.firmware_version)
logger.debug("Built adapter {}".format(adapter))
return getattr(adapter, function_name)
def execute_recorder_command(recorder: Recorder, recorder_command: RecorderCommand, params: Union[dict, None] = None,
connect_by_network_name=True):
if params is None:
params = dict()
if validate_recorder_command_params(recorder_command, params):
try:
func = get_function_from_recorder_command(recorder, recorder_command, connect_by_network_name)
logger.debug(
"Executing func: {} with params: '{}'".format(func, ", ".join(": ".join(str(_)) for _ in params.items())))
out = func(**params)
return True, out
except LrcException as e:
return False, str(e)
else:
logger.info("Could not validate given parameters!")
return False