tobias/lrc-backend
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

added code to initialize db with models and recodres

Browse Source
This commit is contained in:
Tobias Kurze
2019-11-20 12:41:05 +01:00
parent bc1347fe99
commit 60ff5bdeaf
20 changed files with 1658 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
backend/recorder_adapters/__init__.py
View File

@@ -1,10 +1,15 @@
import importlib
import inspect
import pkgutil
import sys
import telnetlib
from abc import ABC, abstractmethod
defined_recorder_adapters = None
# monkey patching of telnet lib
from pprint import pprint
original_read_until = telnetlib.Telnet.read_until
original_write = telnetlib.Telnet.write
@@ -92,6 +97,16 @@ class TelnetAdapter(ABC):
class RecorderAdapter:
def __init__(self, address: str, user: str, password: str):
self.address = address
self.user = user
self.password = password
@classmethod
@abstractmethod
def get_recorder_params(cls) -> dict:
pass
@abstractmethod
def _get_name(self):
pass
@@ -108,13 +123,15 @@ class RecorderAdapter:
pass
def get_defined_recorder_adapters():
def get_defined_recorder_adapters() -> list:
rec_adapters_module = importlib.import_module(".recorder_adapters", package='backend')
rec_adapter_class = getattr(rec_adapters_module, "RecorderAdapter") # needed, otherwise subclass check may fail
models = []
found_packages = list(pkgutil.iter_modules(sys.modules[__name__].__path__))
found_packages = list(pkgutil.iter_modules(rec_adapters_module.__path__))
for f_p in found_packages:
importer = f_p[0]
rec_model_module = importer.find_module(f_p[1]).load_module(f_p[1])
rec_model = {'id': f_p[1], 'name': f_p[1], 'commands': {}}
rec_model = {'id': f_p[1], 'name': f_p[1], 'commands': {}, 'path': rec_model_module.__file__}
if hasattr(rec_model_module, 'RECORDER_MODEL_NAME'):
rec_model['name'] = rec_model_module.RECORDER_MODEL_NAME
if hasattr(rec_model_module, 'REQUIRES_USER'):
@@ -122,7 +139,9 @@ def get_defined_recorder_adapters():
if hasattr(rec_model_module, 'REQUIRES_PW'):
rec_model['requires_password'] = rec_model_module.REQUIRES_PW
for name, obj in inspect.getmembers(rec_model_module, inspect.isclass):
if issubclass(obj, RecorderAdapter):
if issubclass(obj, rec_adapter_class) and name != "RecorderAdapter":
rec_model['id'] = rec_model['id'] + "." + obj.__name__
rec_model['class'] = obj
commands = {}
for method_name, method in inspect.getmembers(obj, predicate=inspect.isfunction):
if len(method_name) > 0 and "_" == method_name[0]:
@@ -141,3 +160,19 @@ def get_defined_recorder_adapters():
rec_model["commands"] = commands
models.append(rec_model)
return models
def get_recorder_adapter_by_id(id: str, **kwargs):
global defined_recorder_adapters
if defined_recorder_adapters is None:
defined_recorder_adapters = get_defined_recorder_adapters()
for rec_adapter in defined_recorder_adapters:
if id in rec_adapter.get('id', '').split("."):
return rec_adapter['class'](**kwargs)
return None
if __name__ == '__main__':
print(get_defined_recorder_adapters())
get_recorder_adapter_by_id('SMP35x', address="172.22.246.207", password="123mzsmp")
exit()

43
backend/recorder_adapters/epiphan_base.py
View File

@@ -1,4 +1,12 @@
# Copyright (c) 2019. Tobias Kurze
"""
This is the recorder adapter implementation for Epiphan Recorders. The following Epiphan recorder models are supported:
- LectureRecorder X2
- LectureRecorder
- VGADVI Recorder
- DVI Broadcaster DL
- DVIRecorderDL
"""
import shutil
import time
from datetime import datetime
@@ -14,39 +22,44 @@ from backend.recorder_adapters import RecorderAdapter
# HOST = "localhost"
from backend.tools.exception_decorator import exception_decorator
RECORDER_MODEL_NAME = "Epiphan Recorder Adapter (for: )"
BASE_URL = "http://172.23.8.102" # Audimax SMP 351
USER = "admin"
PW = "lrgrashof+-"
class EpiphanV1(RecorderAdapter):
def __init__(self, url: str, admin_user: str, admin_pw: str):
if not url.startswith('http'):
url = 'http://' + url
self.url = url
self.user = admin_user
self.password = admin_pw
class Epiphan(RecorderAdapter):
def __init__(self, address: str, user: str, password: str, firmware_version: str = "", **kwargs):
if not address.startswith('http'):
address = 'http://' + address
super().__init__(address, user, password)
self.firmware_version = firmware_version
self.session = requests.Session()
self.session.auth = HTTPBasicAuth(self.user, self.password)
@classmethod
def get_recorder_params(cls) -> dict:
return {'_requires_user': True,
'_requires_password': True}
def _get_name(self):
pass
return RECORDER_MODEL_NAME
def _get_version(self):
pass
@exception_decorator(ConnectionError)
def get_recording_status(self) -> dict:
res = self.session.get(self.url + "/admin/ajax/recorder_status.cgi")
res = self.session.get(self.address + "/admin/ajax/recorder_status.cgi")
if res.ok:
return res.json()
raise LrcException(res.text, res.status_code)
@exception_decorator(ConnectionError)
def get_sysinfo(self) -> dict:
res = self.session.get(self.url + "/ajax/sysinfo.cgi")
res = self.session.get(self.address + "/ajax/sysinfo.cgi")
if res.ok:
return res.json()
raise LrcException(res.text, res.status_code)
@@ -63,13 +76,13 @@ class EpiphanV1(RecorderAdapter):
return self.get_recording_status().get('seconds', None)
def start_recording(self):
res = self.session.get(self.url + "/admin/ajax/start_recorder.cgi")
res = self.session.get(self.address + "/admin/ajax/start_recorder.cgi")
if not res.ok:
raise LrcException(res.text, res.status_code)
time.sleep(2) # just a little bit of waiting time -> it takes a bit for the Epiphan to update its state
def stop_recording(self):
res = self.session.get(self.url + "/admin/ajax/stop_recorder.cgi")
res = self.session.get(self.address + "/admin/ajax/stop_recorder.cgi")
if not res.ok:
raise LrcException(res.text, res.status_code)
time.sleep(4) # just a little bit of waiting time -> it takes a bit for the Epiphan to update its state
@@ -114,7 +127,7 @@ class EpiphanV1(RecorderAdapter):
raise LrcException(str(err))
def get_screenshot(self):
ret = self.session.get(self.url + "/admin/grab_frame.cgi?size=256x192&device=DAV93133.vga&_t=1573471990578",
ret = self.session.get(self.address + "/admin/grab_frame.cgi?size=256x192&device=DAV93133.vga&_t=1573471990578",
stream=True)
print(ret)
@@ -125,7 +138,7 @@ class EpiphanV1(RecorderAdapter):
if __name__ == '__main__':
e = EpiphanV1(BASE_URL, USER, PW)
e = Epiphan(BASE_URL, USER, PW)
try:
# print(e.is_recording())
"""

29
backend/recorder_adapters/extron_smp.py
View File

@@ -1,3 +1,6 @@
"""
Recorder Adapter for SMP
"""
import logging
from backend import LrcException
@@ -6,7 +9,7 @@ from backend.tools.exception_decorator import exception_decorator
logger = logging.getLogger("lrc.recorder_adapters.extron_smp")
RECORDER_MODEL_NAME = "SMP 351 / 352"
RECORDER_MODEL_NAME = "Recorder Adapter for SMP 351 and 352"
VERSION = "0.9.0"
REQUIRES_USER = False
REQUIRES_PW = True
@@ -19,14 +22,14 @@ HOST = "129.13.51.109" # Hertz
USER = "admin"
PW = "123mzsmp"
#PW = "audimaxsmp"
# PW = "audimaxsmp"
PW = "smphertz"
class SMP(TelnetAdapter, RecorderAdapter):
class SMP35x(TelnetAdapter, RecorderAdapter):
def __init__(self, address, password, auto_login=True, **kwargs):
super().__init__(address)
self.admin_pw = password
RecorderAdapter.__init__(self, address, "", password)
TelnetAdapter.__init__(self, address)
if auto_login:
self._login()
@@ -40,7 +43,7 @@ class SMP(TelnetAdapter, RecorderAdapter):
raise LrcException(str(e))
self.tn.read_until("\r\nPassword:")
# password = getpass.getpass()
password = self.admin_pw
password = self.password
self.tn.write(password + "\n\r")
out = self.tn.assert_string_in_output("Login Administrator")
@@ -51,10 +54,10 @@ class SMP(TelnetAdapter, RecorderAdapter):
# TODO: loop until logged in...
logger.warning("Could not login (as admin) with given password! {}".format(self.address))
print("re-enter pw")
self.tn.write(self.admin_pw+"\n\r")
self.tn.write(self.password + "\n\r")
print(self.tn.assert_string_in_output("Login Administrator"))
print("WRONG (admin) password!! Exiting!")
print(self.admin_pw)
print(self.password)
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!")
@@ -725,6 +728,11 @@ class SMP(TelnetAdapter, RecorderAdapter):
some advanced options skipped
"""
@classmethod
def get_recorder_params(cls) -> dict:
return {'_requires_user': False,
'_requires_password': True}
def get_input_hdcp_status(self, input_number: int):
"""
returns:
@@ -783,9 +791,8 @@ class SMP(TelnetAdapter, RecorderAdapter):
return TelnetAdapter._get_response_str(self.tn.read_until_non_empty_line())
def main():
smp = SMP(HOST, PW, True)
smp = SMP35x(HOST, PW, True)
print(smp)
print(smp.get_recording_status())
@@ -833,6 +840,6 @@ def main():
print(smp.unmute_output(2))
print(smp.is_muted(2))
if __name__ == '__main__':
main()