commits
tags
import os
import re
import shutil
import subprocess
from pathlib import Path
import eventlet
from flask import Blueprint, jsonify, request, abort
from werkzeug.security import generate_password_hash, check_password_hash
import db
from auth import login_required
from utils.display import (
sway_env, audio_device_id, parse_profiles,
apply_saved_audio, apply_display_mode,
)
from utils.system import local_ip
bp = Blueprint('settings', __name__, url_prefix='/api')
def _apply_internal_display(enabled, env):
"""Aktivera eller stäng av eDP-1 (intern display) via sway."""
try:
if enabled:
subprocess.run(['swaymsg', 'output', 'eDP-1', 'enable'], env=env, timeout=5)
eventlet.sleep(0.5)
ip = local_ip()
flags = (
'--kiosk --ozone-platform=wayland --noerr --no-first-run '
'--disable-infobars --disable-session-crashed-bubble --disable-translate '
'--disable-web-security --autoplay-policy=no-user-gesture-required '
'--no-sandbox --user-data-dir=/tmp/kiosk-admin'
)
subprocess.run(['swaymsg', 'workspace 2'], env=env, timeout=5)
eventlet.sleep(0.3)
subprocess.run(['swaymsg', 'exec', f'chromium {flags} http://{ip}:5000/admin'],
env=env, timeout=5)
eventlet.sleep(1)
subprocess.run(['swaymsg', 'workspace 1'], env=env, timeout=5)
else:
subprocess.run(['pkill', '-f', 'user-data-dir=/tmp/kiosk-admin'], timeout=5)
eventlet.sleep(0.3)
subprocess.run(['swaymsg', 'output', 'eDP-1', 'disable'], env=env, timeout=5)
except Exception:
pass
@bp.route('/settings/apply_display', methods=['POST'])
@login_required
def apply_display_now():
data = request.get_json(silent=True) or {}
mode = data.get('display_mode')
target = data.get('display_output') or None
if not mode:
state = db.get_state()
mode = state.get('display_mode', '')
if not mode:
return jsonify({'ok': False, 'error': 'no_mode'}), 400
try:
ok = apply_display_mode(str(mode), target_output=target)
apply_saved_audio()
return jsonify({'ok': bool(ok)})
except Exception as e:
return jsonify({'ok': False, 'error': str(e)}), 500
@bp.route('/audio/profiles')
@login_required
def audio_profiles():
dev_id = audio_device_id()
if dev_id is None:
return jsonify([])
try:
pw_env = {**os.environ, 'XDG_RUNTIME_DIR': f'/run/user/{os.getuid()}'}
out = subprocess.check_output(
['pw-cli', 'enum-params', str(dev_id), 'EnumProfile'],
timeout=5, env=pw_env
).decode()
return jsonify(parse_profiles(out))
except Exception:
return jsonify([])
@bp.route('/displays')
@login_required
def list_displays():
outputs = []
try:
for conn in sorted(Path('/sys/class/drm').iterdir()):
status_file = conn / 'status'
if not status_file.exists():
continue
parts = conn.name.split('-', 1)
if len(parts) != 2:
continue
connector = parts[1]
connected = status_file.read_text().strip() == 'connected'
outputs.append({'name': connector, 'connected': connected})
except Exception:
pass
# Best-effort: xrandr-läge (X11)
try:
if shutil.which('xrandr'):
out = ''
for disp in (None, ':0', ':0.0', ':1'):
try:
env = os.environ.copy()
if disp:
env['DISPLAY'] = disp
out = subprocess.check_output(['xrandr', '--query'], env=env, timeout=5).decode(errors='ignore')
if out and ' connected' in out:
break
except Exception:
out = ''
lines = out.splitlines() if out else []
cur_out = None
modes_map = {}
for line in lines:
m = re.match(r"^([A-Za-z0-9-]+) connected", line)
if m:
cur_out = m.group(1)
modes_map[cur_out] = []
continue
if cur_out:
m2 = re.match(r"^\s+(\d+)x(\d+)\s+([0-9\.]+)\s*Hz?(.*)$", line)
if m2:
token = f"{m2.group(1)}x{m2.group(2)}@{m2.group(3)}"
if token not in modes_map[cur_out]:
modes_map[cur_out].append(token)
if '*' in (m2.group(4) or '') or 'current' in (m2.group(4) or ''):
modes_map.setdefault('__current__', {})[cur_out] = token
elif line.strip() == '':
cur_out = None
if modes_map:
for o in outputs:
if o['name'] in modes_map and modes_map[o['name']]:
o['modes'] = modes_map[o['name']]
curmap = modes_map.get('__current__', {})
for o in outputs:
if o['name'] in curmap:
o['current_mode'] = curmap[o['name']]
except Exception:
pass
# Best-effort: wlr-randr-läge (Wayland)
try:
if not any(o.get('modes') for o in outputs) and shutil.which('wlr-randr'):
env = os.environ.copy()
if not env.get('WAYLAND_DISPLAY'):
xdg_runtime = env.get('XDG_RUNTIME_DIR', f'/run/user/{os.getuid()}')
for sock in sorted(Path(xdg_runtime).glob('wayland-[0-9]*')):
if sock.suffix != '.lock':
env['WAYLAND_DISPLAY'] = sock.name
break
out = subprocess.check_output(['wlr-randr'], timeout=5, env=env).decode(errors='ignore')
cur_out = None
modes_map = {}
for line in out.splitlines():
if not line.strip():
continue
if line[0] not in (' ', '\t'):
cur_out = line.strip().split()[0]
modes_map[cur_out] = []
continue
m = re.search(r"(\d+)x(\d+)\s+px,\s+([0-9\.]+)\s+Hz(?:.*(current))?", line)
if cur_out and m:
token = f"{m.group(1)}x{m.group(2)}@{m.group(3)}"
if token not in modes_map[cur_out]:
modes_map[cur_out].append(token)
if 'current' in line:
modes_map.setdefault('__current__', {})[cur_out] = token
current_map = modes_map.pop('__current__', {})
for o in outputs:
if o['name'] in modes_map and modes_map[o['name']]:
o['modes'] = modes_map[o['name']]
if o['name'] in current_map:
o['current_mode'] = current_map[o['name']]
except Exception:
pass
return jsonify(outputs)
@bp.route('/volume', methods=['GET'])
@login_required
def get_volume():
try:
out = subprocess.check_output(['wpctl', 'get-volume', '@DEFAULT_AUDIO_SINK@'], timeout=5).decode()
vol = round(float(out.split()[1]) * 100)
return jsonify({'volume': vol})
except Exception:
return jsonify({'volume': 100})
@bp.route('/volume', methods=['POST'])
@login_required
def set_volume():
data = request.get_json()
vol = max(0, min(100, int(data.get('volume', 100))))
subprocess.run(['wpctl', 'set-volume', '@DEFAULT_AUDIO_SINK@', f'{vol/100:.2f}'], timeout=5)
return '', 204
@bp.route('/settings', methods=['GET'])
@login_required
def get_settings():
state = db.get_state()
return jsonify({
'audio_profile_index': int(state.get('audio_profile_index', 1)),
'display_output': state.get('display_output', 'last'),
'display_mode': state.get('display_mode', ''),
'internal_display': state.get('internal_display', 'on'),
})
@bp.route('/settings', methods=['POST'])
@login_required
def save_settings():
data = request.get_json()
if not data:
abort(400)
profile_index = int(data.get('audio_profile_index', 1))
display = data.get('display_output', 'last')
display_mode = str(data.get('display_mode', '')).strip()
internal_display = 'on' if data.get('internal_display', True) else 'off'
db.set_state('audio_profile_index', str(profile_index))
db.set_state('display_output', display)
db.set_state('internal_display', internal_display)
if display_mode:
db.set_state('display_mode', display_mode)
conf = Path(os.path.dirname(__file__)).parent.parent / 'kiosk.conf'
conf.write_text(
'# Genererad av kiosk-admin — ändra ej manuellt\n'
f'KIOSK_AUDIO_PROFILE_INDEX={profile_index}\n'
f'KIOSK_DISPLAY_OUTPUT={display}\n'
f'KIOSK_INTERNAL_DISPLAY={internal_display}\n'
+ (f'KIOSK_DISPLAY_MODE={display_mode}\n' if display_mode else '')
)
dev_id = audio_device_id()
if dev_id is not None:
try:
pw_env = {**os.environ, 'XDG_RUNTIME_DIR': f'/run/user/{os.getuid()}'}
subprocess.run(
['pw-cli', 's', str(dev_id), 'Profile', f'{{ index: {profile_index} }}'],
timeout=5, check=True, env=pw_env
)
except Exception:
pass
return '', 204
@bp.route('/internal_display', methods=['POST'])
@login_required
def toggle_internal_display():
data = request.get_json()
if data is None:
abort(400)
enabled = bool(data.get('enabled', True))
val = 'on' if enabled else 'off'
db.set_state('internal_display', val)
state = db.get_state()
conf = Path(os.path.dirname(__file__)).parent.parent / 'kiosk.conf'
conf.write_text(
'# Genererad av kiosk-admin — ändra ej manuellt\n'
f'KIOSK_AUDIO_PROFILE_INDEX={state.get("audio_profile_index", 1)}\n'
f'KIOSK_DISPLAY_OUTPUT={state.get("display_output", "HDMI-A-1")}\n'
f'KIOSK_INTERNAL_DISPLAY={val}\n'
+ (f'KIOSK_DISPLAY_MODE={state["display_mode"]}\n' if state.get("display_mode") else '')
)
env = sway_env()
if not env:
return '', 204
eventlet.spawn(_apply_internal_display, enabled, env)
return '', 204
@bp.route('/settings/credentials', methods=['POST'])
@login_required
def change_credentials():
data = request.get_json(silent=True) or {}
current = str(data.get('current_password', ''))
new_user = str(data.get('new_username', '')).strip()[:64]
new_pass = str(data.get('new_password', ''))
if not new_user or not new_pass:
return jsonify({'error': 'Användarnamn och lösenord krävs'}), 400
if len(new_pass) < 4:
return jsonify({'error': 'Lösenordet måste vara minst 4 tecken'}), 400
# Verify current credentials
import app as _app
stored_hash = db.get_state().get('admin_password_hash')
if stored_hash:
if not check_password_hash(stored_hash, current):
return jsonify({'error': 'Fel nuvarande lösenord'}), 403
else:
# Fall back to env/default comparison
if current != _app.KIOSK_PASS:
return jsonify({'error': 'Fel nuvarande lösenord'}), 403
db.set_state('admin_username', new_user)
db.set_state('admin_password_hash', generate_password_hash(new_pass))
return jsonify({'ok': True})