62 lines
2.7 KiB
Python
62 lines
2.7 KiB
Python
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
|