csmith1865/obscreen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 16 Wiki Activity

split main logic in files

Browse Source
This commit is contained in:
jr-k 2024-03-01 01:29:26 +01:00
parent 50846b6592
commit d31c8fe98b
15 changed files with 375 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
.dockerignore Normal file
View File

@ -0,0 +1,12 @@
.idea
*.iws
*.iml
*.ipr
out/
data/uploads/*
!data/uploads/sample.jpg
data/db/*
!data/db/slideshow.json.dist
config.json
*.lock
__pycache__/

2
.gitignore vendored
View File

@ -7,6 +7,6 @@ data/uploads/*
!data/uploads/sample.jpg
data/db/*
!data/db/slideshow.json.dist
config.py
config.json
*.lock
__pycache__/

13
Dockerfile Normal file
View File

@ -0,0 +1,13 @@
FROM python:3.9.17-alpine3.17
RUN apk add --no-cache git chromium
RUN apk add --no-cache --virtual .build-deps git gcc musl-dev \
&& pip install flask pysondb-v2==2.1.0 \
&& apk del .build-deps gcc musl-dev
WORKDIR /app
COPY . .
ENTRYPOINT ["python", "/app/obscreen.py"]

5
config.json.dist Normal file
View File

@ -0,0 +1,5 @@
{
"debug": false,
"reverse_proxy_mode": false,
"lx_file": "/home/pi/.config/lxsession/LXDE-pi/autostart"
}

5
config.py.dist
View File

@ -1,5 +0,0 @@
config = {
"debug": False, # Enable autoreload for html/jinja files
"reverse_proxy_mode": False, # True if you want to use nginx on port 80
"lx_file": '/home/pi/.config/lxsession/LXDE-pi/autostart' # Path to lx autostart file
}

115
obscreen.py
View File

@ -1,117 +1,8 @@
#!/usr/bin/python3
import json
import os
import re
import shutil
import subprocess
import sys
import time
from flask import Flask, send_from_directory
from src.manager.SlideManager import SlideManager
from src.manager.ScreenManager import ScreenManager
from src.manager.VariableManager import VariableManager
from src.controller.PlayerController import PlayerController
from src.controller.SlideshowController import SlideshowController
from src.controller.FleetController import FleetController
from src.controller.SysinfoController import SysinfoController
from src.controller.SettingsController import SettingsController
from config import config
# <config>
variable_manager = VariableManager()
vars = variable_manager.get_variable_map()
screen_manager = ScreenManager()
slide_manager = SlideManager()
PLAYER_URL = 'http://localhost:{}'.format(vars['port'].as_int())
with open('./lang/{}.json'.format(vars['lang'].as_string()), 'r') as file:
LANGDICT = json.load(file)
variable_manager.init(LANGDICT)
# </config>
# <reverse-proxy>
if config['reverse_proxy_mode']:
reverse_proxy_config_file = 'system/nginx-obscreen'
with open(reverse_proxy_config_file, 'r') as file:
content = file.read()
with open(reverse_proxy_config_file, 'w') as file:
file.write(re.sub(r'proxy_pass .*?;', 'proxy_pass {};'.format(PLAYER_URL), content))
PLAYER_URL = 'http://localhost'
# </reverse-proxy>
# <server>
app = Flask(__name__, template_folder='views', static_folder='data')
app.config['UPLOAD_FOLDER'] = 'data/uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB
if config['debug']:
app.config['TEMPLATES_AUTO_RELOAD'] = True
# </server>
# <xenv>
if config['lx_file']:
destination_path = '/home/pi/.config/lxsession/LXDE-pi/autostart'
os.makedirs(os.path.dirname(config['lx_file']), exist_ok=True)
xenv_presets = f"""
@lxpanel --profile LXDE-pi
@pcmanfm --desktop --profile LXDE-pi
@xscreensaver -no-splash
#@point-rpi
@xset s off
@xset -dpms
@xset s noblank
@unclutter -display :0 -noevents -grab
@sed -i 's/"exited_cleanly": false/"exited_cleanly": true/' ~/.config/chromium/Default/Preferences
#@sleep 10
@chromium-browser --disable-features=Translate --ignore-certificate-errors --disable-web-security --disable-restore-session-state --autoplay-policy=no-user-gesture-required --start-maximized --allow-running-insecure-content --remember-cert-error-decisions --disable-restore-session-state --noerrdialogs --kiosk --incognito --window-position=0,0 --display=:0 {PLAYER_URL}
"""
with open(config['lx_file'], 'w') as file:
file.write(xenv_presets)
# </xenv>
# <web>
@app.context_processor
def inject_global_vars():
return dict(
FLEET_ENABLED=vars['fleet_enabled'].as_bool(),
LANG=vars['lang'].as_string(),
STATIC_PREFIX='/data/www/'
)
@app.template_filter('ctime')
def time_ctime(s):
return time.ctime(s)
PlayerController(app, LANGDICT, slide_manager)
SlideshowController(app, LANGDICT, slide_manager, variable_manager)
SettingsController(app, LANGDICT, variable_manager)
SysinfoController(app, LANGDICT, config, variable_manager)
if vars['fleet_enabled'].as_bool():
FleetController(app, LANGDICT, screen_manager)
@app.errorhandler(404)
def not_found(e):
return send_from_directory('views', 'core/error404.html'), 404
# </web>
from src.Application import Application
if __name__ == '__main__':
app.run(
host=vars['bind'].as_string(),
port=vars['port'].as_int(),
debug=config['debug']
)
app = Application(project_dir=os.path.dirname(__file__))
app.start()

28
src/Application.py Normal file
View File

@ -0,0 +1,28 @@
import sys
import logging
import signal
import threading
from src.service.ModelManager import ModelManager
from src.service.WebServer import WebServer
class Application:
def __init__(self, project_dir: str):
self._project_dir = project_dir
self._stop_event = threading.Event()
self._model_manager = ModelManager()
self._web_server = WebServer(project_dir=project_dir, model_manager=self._model_manager)
signal.signal(signal.SIGINT, self.signal_handler)
def start(self) -> None:
self._web_server.run()
def signal_handler(self, signal, frame) -> None:
logging.info("Shutting down...")
self._stop_event.set()
sys.exit(0)

8
src/controller/SysinfoController.py
View File

@ -4,15 +4,17 @@ import platform
import subprocess
from flask import Flask, render_template, jsonify
from src.manager.VariableManager import VariableManager
from src.manager.ConfigManager import ConfigManager
from src.utils import get_ip_address
class SysinfoController:
def __init__(self, app, lang_dict, config, variable_manager):
def __init__(self, app, lang_dict, config_manager: ConfigManager, variable_manager: VariableManager):
self._app = app
self._lang_dict = lang_dict
self._config = config
self._config_manager = config_manager
self._variable_manager = variable_manager
self.register()
@ -32,7 +34,7 @@ class SysinfoController:
def sysinfo_restart(self):
if platform.system().lower() == 'darwin':
if self._config['debug']:
if self._config_manager.map().get('debug'):
python = sys.executable
os.execl(python, python, *sys.argv)
else:

115
src/manager/ConfigManager.py Normal file
View File

@ -0,0 +1,115 @@
import re
import os
import json
import logging
import argparse
from src.manager.VariableManager import VariableManager
class ConfigManager:
CONFIG_FILE = 'config.json'
def __init__(self, variable_manager: VariableManager):
self._variable_manager = variable_manager
self._CONFIG = {
'debug': False,
'reverse_proxy_mode': False,
'lx_file': '/home/pi/.config/lxsession/LXDE-pi/autostart',
'log_file': None,
'log_level': 'INFO',
'log_stdout': True,
'player_url': 'http://localhost:{}'.format(self._variable_manager.map().get('port').as_int())
}
self.load_from_json(file_path=self.CONFIG_FILE)
self.load_from_env()
self.load_from_args()
self.autoconfigure()
if self.map().get('debug'):
logging.debug(self._CONFIG)
def map(self) -> dict:
return self._CONFIG
def parse_arguments(self):
parser = argparse.ArgumentParser(description="Obscreen")
parser.add_argument('--debug', '-d', default=self._CONFIG['debug'], help='Debug mode')
parser.add_argument('--reverse_proxy_mode', '-r', default=self._CONFIG['reverse_proxy_mode'], action='store_true', help='true if you want to use nginx on port 80')
parser.add_argument('--lx-file', '-x', default=self._CONFIG['lx_file'], help='Path to lx autostart file')
parser.add_argument('--log-file', '-lf', default=self._CONFIG['log_file'], help='Log File path')
parser.add_argument('--log-level', '-ll', default=self._CONFIG['log_level'], help='Log Level')
parser.add_argument('--log-stdout', '-ls', default=self._CONFIG['log_stdout'], action='store_true', help='Log to standard output')
return parser.parse_args()
def load_from_args(self) -> None:
args = self.parse_arguments()
if args.debug:
self._CONFIG['debug'] = args.debug
if args.reverse_proxy_mode:
self._CONFIG['reverse_proxy_mode'] = args.reverse_proxy_mode
if args.lx_file:
self._CONFIG['lx_file'] = args.lx_file
if args.log_file:
self._CONFIG['log_file'] = args.log_file
if args.log_level:
self._CONFIG['log_level'] = args.log_level
if args.log_stdout:
self._CONFIG['log_stdout'] = args.log_stdout
def load_from_json(self, file_path: str) -> None:
try:
with open(file_path, 'r') as file:
json_config = json.load(file)
for key in json_config:
self._CONFIG[key] = json_config[key]
logging.info(f"Json var {key} has been found")
except FileNotFoundError:
logging.error(f"Json configuration file {file_path} doesn't exist.")
def load_from_env(self) -> None:
for key in self._CONFIG:
if key in os.environ:
self._CONFIG[key] = os.environ[key]
logging.info(f"Env var {key} has been found")
def autoconfigure(self) -> None:
if self.map().get('reverse_proxy_mode'):
self.autoconfigure_nginx()
if self.map().get('lx_file'):
self.autoconfigure_lxconf()
def autoconfigure_nginx(self) -> None:
reverse_proxy_config_file = 'system/nginx-obscreen'
with open(reverse_proxy_config_file, 'r') as file:
content = file.read()
with open(reverse_proxy_config_file, 'w') as file:
file.write(re.sub(r'proxy_pass .*?;', 'proxy_pass {};'.format(self.map().get('player_url')), content))
self._CONFIG['player_url'] = 'http://localhost'
def autoconfigure_lxconf(self) -> None:
destination_path = self.map().get('lx_file')
player_url = self.map().get('player_url')
os.makedirs(os.path.dirname(destination_path), exist_ok=True)
xenv_presets = f"""
@lxpanel --profile LXDE-pi
@pcmanfm --desktop --profile LXDE-pi
@xscreensaver -no-splash
#@point-rpi
@xset s off
@xset -dpms
@xset s noblank
@unclutter -display :0 -noevents -grab
@sed -i 's/"exited_cleanly": false/"exited_cleanly": true/' ~/.config/chromium/Default/Preferences
#@sleep 10
@chromium-browser --disable-features=Translate --ignore-certificate-errors --disable-web-security --disable-restore-session-state --autoplay-policy=no-user-gesture-required --start-maximized --allow-running-insecure-content --remember-cert-error-decisions --disable-restore-session-state --noerrdialogs --kiosk --incognito --window-position=0,0 --display=:0 {player_url}
"""
with open(destination_path, 'w') as file:
file.write(xenv_presets)

21
src/manager/LangManager.py Normal file
View File

@ -0,0 +1,21 @@
import json
import logging
class LangManager:
LANG_FILE = "lang/{}.json"
def __init__(self, lang: str):
self._map = {}
file_name = self.LANG_FILE.format(lang)
try:
with open(file_name, 'r') as file:
self._map = json.load(file)
except FileNotFoundError:
logging.error("Lang file {} not found".format(file_name))
def map(self) -> dict:
return self._map

34
src/manager/LoggingManager.py Normal file
View File

@ -0,0 +1,34 @@
import sys
import logging
from src.manager.ConfigManager import ConfigManager
class LoggingManager:
def __init__(self, config_manager: ConfigManager):
c_map = config_manager.map()
log_level_str = c_map.get('log_level', 'INFO').upper()
log_level = getattr(logging, log_level_str, logging.INFO)
self._logger = logging.getLogger()
self._logger.setLevel(log_level)
if c_map.get('log_file'):
self._add_file_handler(file_path=c_map.get('log_file'))
if c_map.get('log_stdout'):
self._add_stdout_handler()
def _add_file_handler(self, file_path: str) -> None:
file_handler = logging.FileHandler(file_path)
file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(file_formatter)
self._logger.addHandler(file_handler)
def _add_stdout_handler(self) -> None:
console_handler = logging.StreamHandler(sys.stdout)
console_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler.setFormatter(console_formatter)
self._logger.addHandler(console_handler)

12
src/manager/VariableManager.py
View File

@ -12,9 +12,10 @@ class VariableManager:
def __init__(self):
self._db = PysonDB(self.DB_FILE)
self.init()
self._var_map = {}
self.reload()
def init(self, lang_dict: Optional[Dict] = None) -> None:
def reload(self, lang_dict: Optional[Dict] = None) -> None:
default_vars = [
{"name": "port", "value": 5000, "type": VariableType.INT.value, "editable": True, "description": lang_dict['settings_variable_help_port'] if lang_dict else ""},
{"name": "bind", "value": '0.0.0.0', "type": VariableType.STRING.value, "editable": True, "description": lang_dict['settings_variable_help_bind'] if lang_dict else ""},
@ -37,7 +38,12 @@ class VariableManager:
if variable.name == 'last_restart':
self._db.update_by_id(variable.id, {"value": time.time()})
def get_variable_map(self) -> Dict[str, Variable]:
self._var_map = self.prepare_variable_map()
def map(self) -> dict:
return self._var_map
def prepare_variable_map(self) -> Dict[str, Variable]:
var_map = {}
for var in self.get_all():

39
src/service/ModelManager.py Normal file
View File

@ -0,0 +1,39 @@
from src.manager.SlideManager import SlideManager
from src.manager.ScreenManager import ScreenManager
from src.manager.VariableManager import VariableManager
from src.manager.LangManager import LangManager
from src.manager.ConfigManager import ConfigManager
from src.manager.LoggingManager import LoggingManager
class ModelManager:
def __init__(self):
self._variable_manager = VariableManager()
self._config_manager = ConfigManager(variable_manager=self._variable_manager)
self._logging_manager = LoggingManager(config_manager=self._config_manager)
self._screen_manager = ScreenManager()
self._slide_manager = SlideManager()
self._lang_manager = LangManager(
lang=self.variable().map().get('lang').as_string()
)
self._variable_manager.reload(lang_dict=self._lang_manager.map())
def logging(self) -> LoggingManager:
return self._logging_manager
def config(self) -> ConfigManager:
return self._config_manager
def variable(self) -> VariableManager:
return self._variable_manager
def slide(self) -> SlideManager:
return self._slide_manager
def screen(self) -> ScreenManager:
return self._screen_manager
def lang(self) -> LangManager:
return self._lang_manager

89
src/service/WebServer.py Normal file
View File

@ -0,0 +1,89 @@
import os
import time
from flask import Flask, send_from_directory
from src.service.ModelManager import ModelManager
from src.controller.PlayerController import PlayerController
from src.controller.SlideshowController import SlideshowController
from src.controller.FleetController import FleetController
from src.controller.SysinfoController import SysinfoController
from src.controller.SettingsController import SettingsController
class WebServer:
FOLDER_TEMPLATES = "views"
FOLDER_STATIC = "data"
FOLDER_STATIC_WEB_UPLOADS = "uploads"
FOLDER_STATIC_WEB_ASSETS = "www"
MAX_UPLOADS = 16 * 1024 * 1024
def __init__(self, project_dir: str, model_manager: ModelManager):
self._project_dir = project_dir
self._model_manager = model_manager
self._debug = self._model_manager.config().map().get('debug')
self.setup()
def run(self) -> None:
self._app.run(
host=self._model_manager.variable().map().get('bind').as_string(),
port=self._model_manager.variable().map().get('port').as_int(),
debug=self._debug
)
def setup(self) -> None:
self._setup_flask_app()
self._setup_view_globals()
self._setup_view_extensions()
self._setup_view_errors()
self._setup_view_controllers()
def _get_template_folder(self) -> str:
return "{}/{}".format(self._project_dir, self.FOLDER_TEMPLATES)
def _get_static_folder(self) -> str:
return "{}/{}".format(self._project_dir, self.FOLDER_STATIC)
def _setup_flask_app(self) -> None:
self._app = Flask(
__name__,
template_folder=self._get_template_folder(),
static_folder=self._get_static_folder(),
)
self._app.config['UPLOAD_FOLDER'] = "{}/{}".format(self.FOLDER_STATIC, self.FOLDER_STATIC_WEB_UPLOADS)
self._app.config['MAX_CONTENT_LENGTH'] = self.MAX_UPLOADS
if self._debug:
self._app.config['TEMPLATES_AUTO_RELOAD'] = True
def _setup_view_controllers(self) -> None:
lang_map = self._model_manager.lang().map()
mm = self._model_manager
PlayerController(self._app, lang_map, mm.slide())
SlideshowController(self._app, lang_map, mm.slide(), mm.variable())
SettingsController(self._app, lang_map, mm.variable())
SysinfoController(self._app, lang_map, mm.config(), mm.variable())
if self._model_manager.variable().map().get('fleet_enabled').as_bool():
FleetController(self._app, lang_map, mm.screen())
def _setup_view_globals(self) -> None:
@self._app.context_processor
def inject_global_vars():
return dict(
FLEET_ENABLED=self._model_manager.variable().map().get('fleet_enabled').as_bool(),
LANG=self._model_manager.variable().map().get('lang').as_string(),
STATIC_PREFIX="/{}/{}/".format(self.FOLDER_STATIC, self.FOLDER_STATIC_WEB_ASSETS)
)
def _setup_view_extensions(self) -> None:
@self._app.template_filter('ctime')
def time_ctime(s):
return time.ctime(s)
def _setup_view_errors(self) -> None:
@self._app.errorhandler(404)
def not_found(e):
return send_from_directory(self._get_template_folder(), 'core/error404.html'), 404

2
src/utils.py
View File

@ -1,3 +1,4 @@
import re
import subprocess
import platform
@ -11,7 +12,6 @@ def str_to_enum(str_val: str, enum_class) -> Enum:
return enum_item
raise ValueError(f"{str_val} is not a valid {enum_class.__name__} item")
def get_ip_address() -> Optional[str]:
try:
os_name = platform.system().lower()