mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-17 20:39:13 +00:00
refactor(api-v3): split the 10,469-line blueprint into a package
web_interface/blueprints/api_v3.py held 111 routes, 56 helpers and 181 functions in one module -- 9% of the core by line count and three times the next largest file. It becomes a package of nine route modules grouped by path segment, plus __init__.py for the shared imports, constants, Blueprint and helpers. Every route module decorates the SAME api_v3 Blueprint object, so endpoint names stay api_v3.<function>, the URL map is unchanged and app.py is untouched. Verified: 111 routes before, 111 after, byte-identical rules, endpoints and methods, and every endpoint still on the one blueprint. plugins 3,867 config 1,178 starlark 692 system 619 fonts 452 misc 398 wifi 361 display 326 backup 212 __init__ 1,787 (imports, constants, Blueprint, 56 helpers) Two things the URL-map check could not catch, both found by running the suite: 1. PROJECT_ROOT = Path(__file__).parent.parent.parent. Moving the code one directory deeper made that resolve to web_interface/ instead of the project root. Nothing failed at import; it surfaced as ~110 tests failing with 404s and "installation script not found", because every path built from it was one level too shallow. Now parents[3], and test_api_v3_url_map.py asserts PROJECT_ROOT/run.py exists so the next move cannot repeat it. 2. Module-attribute patching. Tests do monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", ...) and a route module that binds such a name by value never sees the patch. The shared code therefore stays in __init__.py rather than moving to a _common submodule -- it has to live on the module the tests patch -- and the eleven names tests patch are read back through the package (_pkg.X) instead of bound by value. Those eleven were found by AST-scanning every setattr in the test tree, not by guessing; "time" is among them, used to drive a fake clock through the second-resolution credential-backup filenames. Test changes are confined to what genuinely moved: patch targets that now name the owning route module, imports of helpers, and six tests that scan the api_v3 source as a file and now read the package directory. Full suite: 4,278 passed, 68 skipped, 0 failed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
This commit is contained in:
ChuckBuilds
co-authored by
Claude Opus 5
Claude Opus 5
parent
a29c84208e
commit
60d5d19898
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,212 @@
|
||||
"""Backup creation, listing and restore.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
PROJECT_ROOT, Path, _safe_backup_path, api_v3,
|
||||
datetime, json, jsonify, logger, os, plugin_store_manager, request,
|
||||
tempfile,
|
||||
)
|
||||
import web_interface.blueprints.api_v3 as _pkg
|
||||
# Read through the module rather than bound by value: tests patch these
|
||||
# as module attributes, and a value binding would not see the patch.
|
||||
# Several are also called from helpers that live in __init__, so the
|
||||
# package is the only patch point that covers every caller.
|
||||
|
||||
|
||||
@api_v3.route('/backup/preview', methods=['GET'])
|
||||
def backup_preview():
|
||||
"""Return a summary of what a new backup would include."""
|
||||
try:
|
||||
from src.backup_manager import preview_backup_contents
|
||||
data = preview_backup_contents(PROJECT_ROOT)
|
||||
return jsonify({'status': 'success', 'data': data})
|
||||
except Exception as e:
|
||||
logger.error("backup_preview failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
|
||||
@api_v3.route('/backup/list', methods=['GET'])
|
||||
def backup_list():
|
||||
"""List backup ZIPs stored in the export directory."""
|
||||
try:
|
||||
_pkg._BACKUP_EXPORT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
entries = []
|
||||
for p in sorted(_pkg._BACKUP_EXPORT_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True):
|
||||
if not p.is_file() or p.suffix != '.zip':
|
||||
continue
|
||||
st = p.stat()
|
||||
entries.append({
|
||||
'filename': p.name,
|
||||
'size': st.st_size,
|
||||
'created_at': datetime.fromtimestamp(st.st_mtime).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
})
|
||||
return jsonify({'status': 'success', 'data': entries})
|
||||
except Exception as e:
|
||||
logger.error("backup_list failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
|
||||
@api_v3.route('/backup/export', methods=['POST'])
|
||||
def backup_export():
|
||||
"""Create a new backup ZIP and return its filename."""
|
||||
try:
|
||||
from src.backup_manager import create_backup
|
||||
zip_path = create_backup(PROJECT_ROOT, output_dir=_pkg._BACKUP_EXPORT_DIR)
|
||||
return jsonify({'status': 'success', 'filename': zip_path.name})
|
||||
except Exception as e:
|
||||
logger.error("backup_export failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
|
||||
@api_v3.route('/backup/validate', methods=['POST'])
|
||||
def backup_validate():
|
||||
"""Validate an uploaded backup ZIP and return its manifest."""
|
||||
try:
|
||||
from src.backup_manager import validate_backup
|
||||
if 'backup_file' not in request.files:
|
||||
return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400
|
||||
f = request.files['backup_file']
|
||||
with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp:
|
||||
tmp_path = tmp.name
|
||||
f.save(tmp_path)
|
||||
try:
|
||||
ok, err_msg, manifest = validate_backup(Path(tmp_path))
|
||||
finally:
|
||||
try:
|
||||
os.unlink(tmp_path)
|
||||
except OSError:
|
||||
pass
|
||||
if not ok:
|
||||
logger.warning("Backup validation failed: %s", err_msg)
|
||||
return jsonify({'status': 'error', 'message': 'Invalid or corrupted backup file'}), 400
|
||||
safe_manifest = {
|
||||
'schema_version': manifest.get('schema_version'),
|
||||
'created_at': manifest.get('created_at'),
|
||||
'ledmatrix_version': manifest.get('ledmatrix_version'),
|
||||
'hostname': manifest.get('hostname'),
|
||||
'contents': manifest.get('contents', []),
|
||||
'detected_contents': manifest.get('detected_contents', []),
|
||||
'plugins': manifest.get('plugins', []),
|
||||
'total_uncompressed': manifest.get('total_uncompressed'),
|
||||
'file_count': manifest.get('file_count'),
|
||||
}
|
||||
return jsonify({'status': 'success', 'data': safe_manifest})
|
||||
except Exception as e:
|
||||
logger.error("backup_validate failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
|
||||
@api_v3.route('/backup/restore', methods=['POST'])
|
||||
def backup_restore():
|
||||
"""Restore a backup ZIP with optional RestoreOptions."""
|
||||
try:
|
||||
from src.backup_manager import restore_backup, RestoreOptions
|
||||
if 'backup_file' not in request.files:
|
||||
return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400
|
||||
f = request.files['backup_file']
|
||||
options_raw = request.form.get('options', '{}')
|
||||
try:
|
||||
opts_dict = json.loads(options_raw)
|
||||
except json.JSONDecodeError:
|
||||
opts_dict = None
|
||||
if not isinstance(opts_dict, dict):
|
||||
# Every option defaults to True, so falling back to {} on a
|
||||
# parse failure would silently perform a FULL restore —
|
||||
# secrets and all — for a caller who asked for a narrow one
|
||||
# and mis-serialized it. Refuse instead of guessing.
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'Invalid options: expected a JSON object',
|
||||
}), 400
|
||||
options = RestoreOptions(
|
||||
restore_config=bool(opts_dict.get('restore_config', True)),
|
||||
restore_secrets=bool(opts_dict.get('restore_secrets', True)),
|
||||
restore_wifi=bool(opts_dict.get('restore_wifi', True)),
|
||||
restore_fonts=bool(opts_dict.get('restore_fonts', True)),
|
||||
restore_plugin_uploads=bool(opts_dict.get('restore_plugin_uploads', True)),
|
||||
reinstall_plugins=bool(opts_dict.get('reinstall_plugins', True)),
|
||||
)
|
||||
with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp:
|
||||
tmp_path = tmp.name
|
||||
f.save(tmp_path)
|
||||
try:
|
||||
result = restore_backup(Path(tmp_path), PROJECT_ROOT, options)
|
||||
finally:
|
||||
try:
|
||||
os.unlink(tmp_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
# Reinstall plugins if requested and store manager available
|
||||
if options.reinstall_plugins and result.plugins_to_install:
|
||||
psm = getattr(api_v3, 'plugin_store_manager', None) or plugin_store_manager
|
||||
for plug in result.plugins_to_install:
|
||||
pid = plug.get('plugin_id')
|
||||
if not pid:
|
||||
continue
|
||||
try:
|
||||
if psm and hasattr(psm, 'install_plugin'):
|
||||
ok = psm.install_plugin(pid)
|
||||
if ok:
|
||||
result.plugins_installed.append(pid)
|
||||
else:
|
||||
result.plugins_failed.append({'plugin_id': pid, 'error': 'install_plugin returned False'})
|
||||
else:
|
||||
result.plugins_failed.append({'plugin_id': pid, 'error': 'Store manager unavailable'})
|
||||
except Exception as pe:
|
||||
logger.error(
|
||||
"[Backup] Failed to reinstall plugin %r: %s", pid, pe, exc_info=True
|
||||
)
|
||||
result.plugins_failed.append({'plugin_id': pid, 'error': 'Installation failed; see server logs'})
|
||||
|
||||
# A restore that dropped files can still report success if the only
|
||||
# failures were plugin reinstalls, since those don't touch result.errors.
|
||||
if result.plugins_failed:
|
||||
result.success = False
|
||||
|
||||
data = result.to_dict()
|
||||
if not result.success:
|
||||
# Name what failed, and what nonetheless landed. A restore is
|
||||
# partial far more often than it is total -- a fresh install can
|
||||
# leave config_secrets.json unwritable by the web service, so
|
||||
# config restores and secrets do not. "Restore had errors" alone
|
||||
# left the user unable to tell a wholly failed restore from one
|
||||
# that quietly dropped their API keys.
|
||||
failed_plugins = [
|
||||
str(p.get('plugin_id')) for p in (result.plugins_failed or []) if p.get('plugin_id')
|
||||
]
|
||||
parts = []
|
||||
if result.restored:
|
||||
parts.append(f"restored: {', '.join(result.restored)}")
|
||||
if result.errors:
|
||||
parts.append(f"failed: {'; '.join(result.errors)}")
|
||||
if failed_plugins:
|
||||
parts.append(f"plugins not reinstalled: {', '.join(failed_plugins)}")
|
||||
message = 'Restore incomplete — ' + ('. '.join(parts) if parts else 'see logs')
|
||||
return jsonify({'status': 'error', 'message': message, 'data': data}), 500
|
||||
return jsonify({'status': 'success', 'data': data})
|
||||
except Exception as e:
|
||||
logger.error("backup_restore failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
|
||||
@api_v3.route('/backup/download/<path:filename>', methods=['GET'])
|
||||
def backup_download(filename):
|
||||
"""Stream a backup ZIP to the browser."""
|
||||
from flask import send_from_directory
|
||||
if _safe_backup_path(filename) is None:
|
||||
return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
|
||||
try:
|
||||
# send_from_directory uses werkzeug safe_join internally — CodeQL-recognized sanitizer.
|
||||
return send_from_directory(_pkg._BACKUP_EXPORT_DIR, filename, as_attachment=True)
|
||||
except FileNotFoundError:
|
||||
return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
|
||||
@api_v3.route('/backup/<path:filename>', methods=['DELETE'])
|
||||
def backup_delete(filename):
|
||||
"""Delete a stored backup ZIP."""
|
||||
safe = _safe_backup_path(filename)
|
||||
if safe is None:
|
||||
return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
|
||||
# Enumerate the export directory and match by name so the unlink target is
|
||||
# a filesystem-derived path rather than one constructed from user input.
|
||||
try:
|
||||
for entry in _pkg._BACKUP_EXPORT_DIR.iterdir():
|
||||
if entry.is_file() and entry.name == safe.name:
|
||||
entry.unlink()
|
||||
return jsonify({'status': 'success'})
|
||||
except OSError as e:
|
||||
logger.error("backup_delete failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500
|
||||
return jsonify({'status': 'error', 'message': 'Backup not found'}), 404
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,326 @@
|
||||
"""Display control, on-demand playback and preview.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
_ensure_cache_manager, _ensure_display_service_running,
|
||||
_get_display_service_status, _stop_display_service, api_v3,
|
||||
describe_exception, jsonify, logger, os, request, uuid,
|
||||
)
|
||||
import web_interface.blueprints.api_v3 as _pkg
|
||||
# Read through the module rather than bound by value: tests patch these
|
||||
# as module attributes, and a value binding would not see the patch.
|
||||
# Several are also called from helpers that live in __init__, so the
|
||||
# package is the only patch point that covers every caller.
|
||||
|
||||
|
||||
@api_v3.route('/display/current', methods=['GET'])
|
||||
def get_display_current():
|
||||
"""Get current display state"""
|
||||
try:
|
||||
import base64
|
||||
from PIL import Image
|
||||
import io
|
||||
|
||||
snapshot_path = "/tmp/led_matrix_preview.png"
|
||||
|
||||
# Get display dimensions from config
|
||||
try:
|
||||
if api_v3.config_manager:
|
||||
main_config = api_v3.config_manager.load_config()
|
||||
hardware_config = main_config.get('display', {}).get('hardware', {})
|
||||
cols = hardware_config.get('cols', 64)
|
||||
chain_length = hardware_config.get('chain_length', 2)
|
||||
rows = hardware_config.get('rows', 32)
|
||||
parallel = hardware_config.get('parallel', 1)
|
||||
width = cols * chain_length
|
||||
height = rows * parallel
|
||||
else:
|
||||
width = 128
|
||||
height = 64
|
||||
except Exception:
|
||||
width = 128
|
||||
height = 64
|
||||
|
||||
# Try to read snapshot file
|
||||
image_data = None
|
||||
if os.path.exists(snapshot_path):
|
||||
try:
|
||||
with Image.open(snapshot_path) as img:
|
||||
# Convert to PNG and encode as base64
|
||||
buffer = io.BytesIO()
|
||||
img.save(buffer, format='PNG')
|
||||
image_data = base64.b64encode(buffer.getvalue()).decode('utf-8')
|
||||
except Exception as img_err:
|
||||
# File might be being written or corrupted, return None
|
||||
pass
|
||||
|
||||
display_data = {
|
||||
'timestamp': _pkg.time.time(),
|
||||
'width': width,
|
||||
'height': height,
|
||||
'image': image_data # Base64 encoded image data or None if unavailable
|
||||
}
|
||||
return jsonify({'status': 'success', 'data': display_data})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/display/modes', methods=['GET'])
|
||||
def get_display_modes():
|
||||
"""Every display mode that can be requested on-demand, with its plugin.
|
||||
|
||||
/plugins/installed carries no mode information, so anything driving the
|
||||
display from outside the web UI (the Home Assistant MQTT bridge, a script)
|
||||
had to read each plugin's manifest.json off disk and reimplement the
|
||||
fallbacks in PluginManager.get_plugin_display_modes to do it. This is the
|
||||
same list the force-display dialog offers, from the source that owns it.
|
||||
|
||||
Knowing each mode's plugin_id also matters because /display/on-demand/start
|
||||
falls back to find_plugin_for_mode when plugin_id is omitted, and that
|
||||
lookup only sees modes declared in a static manifest -- a plugin whose
|
||||
modes are generated (each installed Starlark app is one) 404s there.
|
||||
Sending the plugin_id from this list skips the lookup entirely.
|
||||
|
||||
Query params:
|
||||
include_disabled: '1' to list modes of disabled plugins too. They can
|
||||
still be requested on-demand -- the controller enables the plugin
|
||||
for the duration -- so they are reported with enabled: false
|
||||
rather than omitted.
|
||||
"""
|
||||
try:
|
||||
if not api_v3.plugin_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Plugin manager not initialized'}), 500
|
||||
|
||||
# Discovery is lazy and normally triggered by whichever endpoint runs
|
||||
# first, which is a person opening the dashboard. A caller that never
|
||||
# visits it would otherwise see an empty list.
|
||||
api_v3.plugin_manager.discover_plugins()
|
||||
|
||||
include_disabled = request.args.get('include_disabled') in ('1', 'true', 'True')
|
||||
full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
|
||||
|
||||
modes = []
|
||||
for plugin_id, manifest in sorted(api_v3.plugin_manager.plugin_manifests.items()):
|
||||
# A hand-edited or migrated config.json can hold a non-dict under a
|
||||
# plugin id; DisplayController._reconcile guards the same shape, so
|
||||
# it happens in practice. Without this, .get() raises AttributeError,
|
||||
# the loop aborts and the endpoint answers 500 with no modes at all
|
||||
# -- one bad section would blank every entity the MQTT bridge builds
|
||||
# from this list.
|
||||
plugin_config = full_config.get(plugin_id)
|
||||
if not isinstance(plugin_config, dict):
|
||||
if plugin_config is not None:
|
||||
logger.warning(
|
||||
"Config for plugin %r is %s, not an object; treating it as disabled",
|
||||
plugin_id, type(plugin_config).__name__)
|
||||
plugin_config = {}
|
||||
enabled = bool(plugin_config.get('enabled', False))
|
||||
if not enabled and not include_disabled:
|
||||
continue
|
||||
plugin_name = (manifest or {}).get('name') or plugin_id
|
||||
plugin_modes = api_v3.plugin_manager.get_plugin_display_modes(plugin_id) or [plugin_id]
|
||||
for mode in plugin_modes:
|
||||
# A single-mode plugin's mode is the plugin, so its own name is
|
||||
# the readable label. Multi-mode plugins have no per-mode name
|
||||
# anywhere, so the raw mode string is the only thing to show.
|
||||
modes.append({
|
||||
'mode': mode,
|
||||
'plugin_id': plugin_id,
|
||||
'plugin_name': plugin_name,
|
||||
'name': plugin_name if len(plugin_modes) == 1 else mode,
|
||||
'enabled': enabled,
|
||||
})
|
||||
|
||||
return jsonify({'status': 'success', 'data': {'modes': modes}})
|
||||
except Exception as exc:
|
||||
# describe_exception, not a bare message: test_web_error_detail.py
|
||||
# enforces that every handler here returns it, because a device whose
|
||||
# storage is failing otherwise answers "see logs for details" from the
|
||||
# log viewer too. It redacts credentials out of the exception text.
|
||||
# CodeQL flags this as stack-trace exposure across all ~75 handlers;
|
||||
# it is the project's deliberate, reviewed trade-off.
|
||||
logger.error('Error in get_display_modes', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
|
||||
@api_v3.route('/display/on-demand/status', methods=['GET'])
|
||||
def get_on_demand_status():
|
||||
"""Return the current on-demand display state."""
|
||||
try:
|
||||
cache = _ensure_cache_manager()
|
||||
state = cache.get('display_on_demand_state', max_age=120)
|
||||
if state is None:
|
||||
state = {
|
||||
'active': False,
|
||||
'status': 'idle',
|
||||
'last_updated': None
|
||||
}
|
||||
service_status = _get_display_service_status()
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'state': state,
|
||||
'service': service_status
|
||||
}
|
||||
})
|
||||
except Exception as exc:
|
||||
logger.error('Error in get_on_demand_status', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
|
||||
@api_v3.route('/display/on-demand/start', methods=['POST'])
|
||||
def start_on_demand_display():
|
||||
"""Request the display controller to run a specific plugin on-demand."""
|
||||
try:
|
||||
data = request.get_json(silent=True) or {}
|
||||
plugin_id = data.get('plugin_id')
|
||||
mode = data.get('mode')
|
||||
duration = data.get('duration')
|
||||
pinned = bool(data.get('pinned', False))
|
||||
start_service = data.get('start_service', True)
|
||||
|
||||
if not plugin_id and not mode:
|
||||
return jsonify({'status': 'error', 'message': 'plugin_id or mode is required'}), 400
|
||||
|
||||
resolved_plugin = plugin_id
|
||||
resolved_mode = mode
|
||||
|
||||
if api_v3.plugin_manager:
|
||||
if resolved_plugin and resolved_plugin not in api_v3.plugin_manager.plugin_manifests:
|
||||
return jsonify({'status': 'error', 'message': f'Plugin {resolved_plugin} not found'}), 404
|
||||
|
||||
if resolved_plugin and not resolved_mode:
|
||||
modes = api_v3.plugin_manager.get_plugin_display_modes(resolved_plugin)
|
||||
resolved_mode = modes[0] if modes else resolved_plugin
|
||||
elif resolved_mode and not resolved_plugin:
|
||||
resolved_plugin = api_v3.plugin_manager.find_plugin_for_mode(resolved_mode)
|
||||
if not resolved_plugin:
|
||||
return jsonify({'status': 'error', 'message': f'Mode {resolved_mode} not found'}), 404
|
||||
|
||||
# Note: On-demand can work with disabled plugins - the display controller
|
||||
# will temporarily enable them during initialization if needed
|
||||
# We don't block the request here, but log it for debugging
|
||||
if api_v3.config_manager and resolved_plugin:
|
||||
config = api_v3.config_manager.load_config()
|
||||
plugin_config = config.get(resolved_plugin, {})
|
||||
if 'enabled' in plugin_config and not plugin_config.get('enabled', False):
|
||||
logger.info(
|
||||
"On-demand request for disabled plugin '%s' - will be temporarily enabled",
|
||||
resolved_plugin,
|
||||
)
|
||||
|
||||
# Set the on-demand request in cache FIRST (before starting service)
|
||||
# This ensures the request is available when the service starts/restarts
|
||||
cache = _ensure_cache_manager()
|
||||
request_id = data.get('request_id') or str(uuid.uuid4())
|
||||
request_payload = {
|
||||
'request_id': request_id,
|
||||
'action': 'start',
|
||||
'plugin_id': resolved_plugin,
|
||||
'mode': resolved_mode,
|
||||
'duration': duration,
|
||||
'pinned': pinned,
|
||||
'timestamp': _pkg.time.time()
|
||||
}
|
||||
cache.set('display_on_demand_request', request_payload)
|
||||
|
||||
# Check if display service is running (or will be started)
|
||||
service_status = _get_display_service_status()
|
||||
service_was_running = service_status.get('active', False)
|
||||
|
||||
# Stop the display service first to ensure clean state when we will restart it
|
||||
if service_was_running and start_service:
|
||||
import _pkg.time as time_module
|
||||
logger.debug("Stopping display service before starting on-demand mode")
|
||||
_stop_display_service()
|
||||
# Wait a brief moment for the service to fully stop
|
||||
time_module.sleep(1.5)
|
||||
logger.debug("Display service stopped, now starting with on-demand request")
|
||||
|
||||
if not service_status.get('active') and not start_service:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'Display service is not running. Please start the display service or enable "Start Service" option.',
|
||||
'service_status': service_status
|
||||
}), 400
|
||||
|
||||
service_result = None
|
||||
if start_service:
|
||||
service_result = _ensure_display_service_running()
|
||||
# Check if service actually started
|
||||
if service_result and not service_result.get('active'):
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'Failed to start display service. Please check service logs or start it manually.',
|
||||
'service_result': service_result
|
||||
}), 500
|
||||
|
||||
# Service was restarted (or started fresh) with on-demand request in cache
|
||||
# The display controller will read the request during initialization or when it polls
|
||||
|
||||
response_data = {
|
||||
'request_id': request_id,
|
||||
'plugin_id': resolved_plugin,
|
||||
'mode': resolved_mode,
|
||||
'duration': duration,
|
||||
'pinned': pinned,
|
||||
'service': service_result
|
||||
}
|
||||
return jsonify({'status': 'success', 'data': response_data})
|
||||
except Exception as exc:
|
||||
logger.error('Error in start_on_demand_display', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
|
||||
@api_v3.route('/display/on-demand/stop', methods=['POST'])
|
||||
def stop_on_demand_display():
|
||||
"""Request the display controller to stop on-demand mode."""
|
||||
try:
|
||||
data = request.get_json(silent=True) or {}
|
||||
stop_service = data.get('stop_service', False)
|
||||
|
||||
# Set the stop request in cache FIRST
|
||||
# The display controller will poll this and restart without the on-demand filter
|
||||
cache = _ensure_cache_manager()
|
||||
request_id = data.get('request_id') or str(uuid.uuid4())
|
||||
request_payload = {
|
||||
'request_id': request_id,
|
||||
'action': 'stop',
|
||||
'timestamp': _pkg.time.time()
|
||||
}
|
||||
cache.set('display_on_demand_request', request_payload)
|
||||
|
||||
# Note: The display controller's _clear_on_demand() will handle the restart
|
||||
# to restore normal operation with all plugins
|
||||
|
||||
service_result = None
|
||||
if stop_service:
|
||||
service_result = _stop_display_service()
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'request_id': request_id,
|
||||
'service': service_result
|
||||
}
|
||||
})
|
||||
except Exception as exc:
|
||||
logger.error('Error in stop_on_demand_display', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(exc)}), 500
|
||||
@api_v3.route('/display/current-status', methods=['GET'])
|
||||
def get_current_display_status():
|
||||
"""Return the display mode/plugin currently intended to be shown.
|
||||
|
||||
Published by the display process (display_controller._publish_current_mode_state)
|
||||
to the shared cache whenever the active mode changes, so the web UI (e.g. the
|
||||
System Logs page) can show what's on screen without querying the display
|
||||
process directly.
|
||||
"""
|
||||
try:
|
||||
cache = _ensure_cache_manager()
|
||||
state = cache.get('display_current_state', max_age=120)
|
||||
if state is None:
|
||||
state = {
|
||||
'mode': None,
|
||||
'plugin_id': None,
|
||||
'last_updated': None,
|
||||
}
|
||||
return jsonify({'status': 'success', 'data': state})
|
||||
except Exception as e:
|
||||
logger.error('Error in get_current_display_status', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@@ -0,0 +1,452 @@
|
||||
"""Font catalogue, upload, preview and deletion.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
PROJECT_ROOT, Path, Response, SYSTEM_FONTS, api_v3, describe_exception,
|
||||
jsonify, logger, os, re, request, validate_file_upload,
|
||||
)
|
||||
|
||||
|
||||
@api_v3.route('/fonts/catalog', methods=['GET'])
|
||||
def get_fonts_catalog():
|
||||
"""Get fonts catalog"""
|
||||
try:
|
||||
# Check cache first (5 minute TTL)
|
||||
try:
|
||||
from web_interface.cache import get_cached, set_cached
|
||||
cached_result = get_cached('fonts_catalog', ttl_seconds=300)
|
||||
if cached_result is not None:
|
||||
return jsonify({'status': 'success', 'data': {'catalog': cached_result}})
|
||||
except ImportError:
|
||||
# Cache not available, continue without caching
|
||||
get_cached = None
|
||||
set_cached = None
|
||||
|
||||
# Try to import freetype, but continue without it if unavailable
|
||||
try:
|
||||
import freetype
|
||||
freetype_available = True
|
||||
except ImportError:
|
||||
freetype_available = False
|
||||
|
||||
# Scan assets/fonts directory for actual font files
|
||||
fonts_dir = PROJECT_ROOT / "assets" / "fonts"
|
||||
catalog = {}
|
||||
|
||||
if fonts_dir.exists() and fonts_dir.is_dir():
|
||||
for filename in os.listdir(fonts_dir):
|
||||
if filename.endswith(('.ttf', '.otf', '.bdf')):
|
||||
filepath = fonts_dir / filename
|
||||
# Generate family name from filename (without extension)
|
||||
family_name = os.path.splitext(filename)[0]
|
||||
|
||||
# Try to get font metadata using freetype (for TTF/OTF)
|
||||
metadata = {}
|
||||
if filename.endswith(('.ttf', '.otf')) and freetype_available:
|
||||
try:
|
||||
face = freetype.Face(str(filepath))
|
||||
if face.valid:
|
||||
# Get font family name from font file
|
||||
family_name_from_font = face.family_name.decode('utf-8') if face.family_name else family_name
|
||||
metadata = {
|
||||
'family': family_name_from_font,
|
||||
'style': face.style_name.decode('utf-8') if face.style_name else 'Regular',
|
||||
'num_glyphs': face.num_glyphs,
|
||||
'units_per_em': face.units_per_EM
|
||||
}
|
||||
# Use font's family name if available
|
||||
if family_name_from_font:
|
||||
family_name = family_name_from_font
|
||||
except Exception:
|
||||
# If freetype fails, use filename-based name
|
||||
pass
|
||||
|
||||
# Store relative path from project root
|
||||
relative_path = str(filepath.relative_to(PROJECT_ROOT))
|
||||
font_type = 'ttf' if filename.endswith('.ttf') else 'otf' if filename.endswith('.otf') else 'bdf'
|
||||
|
||||
# Generate human-readable display name from family_name
|
||||
display_name = family_name.replace('-', ' ').replace('_', ' ')
|
||||
# Add space before capital letters for camelCase names
|
||||
display_name = re.sub(r'([a-z])([A-Z])', r'\1 \2', display_name)
|
||||
# Add space before numbers that follow letters
|
||||
display_name = re.sub(r'([a-zA-Z])(\d)', r'\1 \2', display_name)
|
||||
# Clean up multiple spaces
|
||||
display_name = ' '.join(display_name.split())
|
||||
|
||||
# Use filename (without extension) as unique key to avoid collisions
|
||||
# when multiple files share the same family_name from font metadata
|
||||
catalog_key = os.path.splitext(filename)[0]
|
||||
|
||||
# Check if this is a system font (cannot be deleted)
|
||||
is_system = catalog_key.lower() in SYSTEM_FONTS
|
||||
|
||||
catalog[catalog_key] = {
|
||||
'filename': filename,
|
||||
'family_name': family_name,
|
||||
'display_name': display_name,
|
||||
'path': relative_path,
|
||||
'type': font_type,
|
||||
'is_system': is_system,
|
||||
'metadata': metadata if metadata else None
|
||||
}
|
||||
|
||||
# Cache the result (5 minute TTL) if available
|
||||
if set_cached:
|
||||
try:
|
||||
set_cached('fonts_catalog', catalog, ttl_seconds=300)
|
||||
except Exception:
|
||||
logger.error("[FontCatalog] Failed to cache fonts_catalog", exc_info=True)
|
||||
|
||||
return jsonify({'status': 'success', 'data': {'catalog': catalog}})
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/tokens', methods=['GET'])
|
||||
def get_font_tokens():
|
||||
"""Get font size tokens"""
|
||||
try:
|
||||
# This would integrate with the actual font system
|
||||
# For now, return sample tokens
|
||||
tokens = {
|
||||
'xs': 6,
|
||||
'sm': 8,
|
||||
'md': 10,
|
||||
'lg': 12,
|
||||
'xl': 14,
|
||||
'xxl': 16
|
||||
}
|
||||
return jsonify({'status': 'success', 'data': {'tokens': tokens}})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/overrides', methods=['GET'])
|
||||
def get_fonts_overrides():
|
||||
"""Get font overrides"""
|
||||
try:
|
||||
# This would integrate with the actual font system
|
||||
# For now, return empty overrides
|
||||
overrides = {}
|
||||
return jsonify({'status': 'success', 'data': {'overrides': overrides}})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/overrides', methods=['POST'])
|
||||
def save_fonts_overrides():
|
||||
"""Save font overrides"""
|
||||
try:
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
|
||||
|
||||
# This would integrate with the actual font system
|
||||
return jsonify({'status': 'success', 'message': 'Font overrides saved'})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/overrides/<element_key>', methods=['DELETE'])
|
||||
def delete_font_override(element_key):
|
||||
"""Delete font override"""
|
||||
try:
|
||||
# This would integrate with the actual font system
|
||||
return jsonify({'status': 'success', 'message': f'Font override for {element_key} deleted'})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/upload', methods=['POST'])
|
||||
def upload_font():
|
||||
"""Upload font file"""
|
||||
try:
|
||||
if 'font_file' not in request.files:
|
||||
return jsonify({'status': 'error', 'message': 'No font file provided'}), 400
|
||||
|
||||
font_file = request.files['font_file']
|
||||
if font_file.filename == '':
|
||||
return jsonify({'status': 'error', 'message': 'No file selected'}), 400
|
||||
|
||||
# Validate filename
|
||||
is_valid, error_msg = validate_file_upload(
|
||||
font_file.filename,
|
||||
max_size_mb=10,
|
||||
allowed_extensions=['.ttf', '.otf', '.bdf']
|
||||
)
|
||||
if not is_valid:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
font_family = request.form.get('font_family', '')
|
||||
|
||||
if not font_family:
|
||||
return jsonify({'status': 'error', 'message': 'Font file and family name required'}), 400
|
||||
|
||||
# Validate font family name
|
||||
if not font_family.replace('_', '').replace('-', '').isalnum():
|
||||
return jsonify({'status': 'error', 'message': 'Font family name must contain only letters, numbers, underscores, and hyphens'}), 400
|
||||
|
||||
# Save the font file to assets/fonts directory
|
||||
fonts_dir = PROJECT_ROOT / "assets" / "fonts"
|
||||
fonts_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create filename from family name
|
||||
original_ext = os.path.splitext(font_file.filename)[1].lower()
|
||||
safe_filename = f"{font_family}{original_ext}"
|
||||
filepath = fonts_dir / safe_filename
|
||||
|
||||
# Check if file already exists
|
||||
if filepath.exists():
|
||||
return jsonify({'status': 'error', 'message': f'Font with name {font_family} already exists'}), 400
|
||||
|
||||
# Save the file
|
||||
font_file.save(str(filepath))
|
||||
|
||||
# Clear font catalog cache
|
||||
try:
|
||||
from web_interface.cache import delete_cached
|
||||
delete_cached('fonts_catalog')
|
||||
except ImportError as e:
|
||||
logger.warning("[FontUpload] Cache module not available: %s", e)
|
||||
except Exception:
|
||||
logger.error("[FontUpload] Failed to clear fonts_catalog cache", exc_info=True)
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': f'Font {font_family} uploaded successfully',
|
||||
'font_family': font_family,
|
||||
'filename': safe_filename,
|
||||
'path': f'assets/fonts/{safe_filename}'
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/preview', methods=['GET'])
|
||||
def get_font_preview() -> tuple[Response, int] | Response:
|
||||
"""Generate a preview image of text rendered with a specific font"""
|
||||
try:
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
import io
|
||||
import base64
|
||||
|
||||
# Limits to prevent DoS via large image generation on constrained devices
|
||||
MAX_TEXT_CHARS = 100
|
||||
MAX_TEXT_LINES = 3
|
||||
MAX_DIM = 1024 # Max width or height in pixels
|
||||
MAX_PIXELS = 500000 # Max total pixels (e.g., ~700x700)
|
||||
|
||||
font_filename = request.args.get('font', '')
|
||||
text = request.args.get('text', 'Sample Text 123')
|
||||
bg_color = request.args.get('bg', '000000')
|
||||
fg_color = request.args.get('fg', 'ffffff')
|
||||
|
||||
# Validate text length and line count early
|
||||
if len(text) > MAX_TEXT_CHARS:
|
||||
return jsonify({'status': 'error', 'message': f'Text exceeds maximum length of {MAX_TEXT_CHARS} characters'}), 400
|
||||
if text.count('\n') >= MAX_TEXT_LINES:
|
||||
return jsonify({'status': 'error', 'message': f'Text exceeds maximum of {MAX_TEXT_LINES} lines'}), 400
|
||||
|
||||
# Safe integer parsing for size
|
||||
try:
|
||||
size = int(request.args.get('size', 12))
|
||||
except (ValueError, TypeError, OverflowError):
|
||||
return jsonify({'status': 'error', 'message': 'Invalid font size'}), 400
|
||||
|
||||
if not font_filename:
|
||||
return jsonify({'status': 'error', 'message': 'Font filename required'}), 400
|
||||
|
||||
# Validate size
|
||||
if size < 4 or size > 72:
|
||||
return jsonify({'status': 'error', 'message': 'Font size must be between 4 and 72'}), 400
|
||||
|
||||
# Security: Validate font_filename to prevent path traversal
|
||||
# Only allow alphanumeric, hyphen, underscore, and dot (for extension)
|
||||
safe_name = Path(font_filename).name # Strip any directory components
|
||||
if safe_name != font_filename or '..' in font_filename:
|
||||
return jsonify({'status': 'error', 'message': 'Invalid font filename'}), 400
|
||||
|
||||
# Validate extension
|
||||
allowed_extensions = ['.ttf', '.otf', '.bdf']
|
||||
has_valid_ext = any(safe_name.lower().endswith(ext) for ext in allowed_extensions)
|
||||
name_without_ext = safe_name.rsplit('.', 1)[0] if '.' in safe_name else safe_name
|
||||
|
||||
# Find the font file
|
||||
fonts_dir = PROJECT_ROOT / "assets" / "fonts"
|
||||
if not fonts_dir.exists():
|
||||
return jsonify({'status': 'error', 'message': 'Fonts directory not found'}), 404
|
||||
|
||||
font_path = fonts_dir / safe_name
|
||||
|
||||
if not font_path.exists() and not has_valid_ext:
|
||||
# Try finding by family name (without extension)
|
||||
for ext in allowed_extensions:
|
||||
potential_path = fonts_dir / f"{name_without_ext}{ext}"
|
||||
if potential_path.exists():
|
||||
font_path = potential_path
|
||||
break
|
||||
|
||||
# Final security check: ensure path is within fonts_dir
|
||||
try:
|
||||
font_path.resolve().relative_to(fonts_dir.resolve())
|
||||
except ValueError:
|
||||
return jsonify({'status': 'error', 'message': 'Invalid font path'}), 400
|
||||
|
||||
if not font_path.exists():
|
||||
return jsonify({'status': 'error', 'message': f'Font file not found: {font_filename}'}), 404
|
||||
|
||||
# Parse colors
|
||||
try:
|
||||
bg_rgb = tuple(int(bg_color[i:i+2], 16) for i in (0, 2, 4))
|
||||
fg_rgb = tuple(int(fg_color[i:i+2], 16) for i in (0, 2, 4))
|
||||
except (ValueError, IndexError):
|
||||
bg_rgb = (0, 0, 0)
|
||||
fg_rgb = (255, 255, 255)
|
||||
|
||||
# Load font
|
||||
font = None
|
||||
if str(font_path).endswith('.bdf'):
|
||||
# BDF fonts require complex per-glyph rendering via freetype
|
||||
# Return explicit error rather than showing misleading preview with default font
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'BDF font preview not supported. BDF fonts will render correctly on the LED matrix.'
|
||||
}), 400
|
||||
else:
|
||||
# TTF/OTF fonts
|
||||
try:
|
||||
font = ImageFont.truetype(str(font_path), size)
|
||||
except (IOError, OSError) as e:
|
||||
# IOError/OSError raised for invalid/corrupt font files
|
||||
logger.warning("[FontPreview] Failed to load font %s: %s", font_path, e)
|
||||
font = ImageFont.load_default()
|
||||
|
||||
# Calculate text size
|
||||
temp_img = Image.new('RGB', (1, 1))
|
||||
temp_draw = ImageDraw.Draw(temp_img)
|
||||
bbox = temp_draw.textbbox((0, 0), text, font=font)
|
||||
text_width = bbox[2] - bbox[0]
|
||||
text_height = bbox[3] - bbox[1]
|
||||
|
||||
# Create image with padding
|
||||
padding = 10
|
||||
img_width = max(text_width + padding * 2, 100)
|
||||
img_height = max(text_height + padding * 2, 30)
|
||||
|
||||
# Validate resulting image size to prevent memory/CPU spikes
|
||||
if img_width > MAX_DIM or img_height > MAX_DIM:
|
||||
return jsonify({'status': 'error', 'message': 'Requested image too large'}), 400
|
||||
if img_width * img_height > MAX_PIXELS:
|
||||
return jsonify({'status': 'error', 'message': 'Requested image too large'}), 400
|
||||
|
||||
img = Image.new('RGB', (img_width, img_height), bg_rgb)
|
||||
draw = ImageDraw.Draw(img)
|
||||
|
||||
# Center text
|
||||
x = (img_width - text_width) // 2
|
||||
y = (img_height - text_height) // 2
|
||||
|
||||
draw.text((x, y), text, font=font, fill=fg_rgb)
|
||||
|
||||
# Convert to base64
|
||||
buffer = io.BytesIO()
|
||||
img.save(buffer, format='PNG')
|
||||
buffer.seek(0)
|
||||
img_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'image': f'data:image/png;base64,{img_base64}',
|
||||
'width': img_width,
|
||||
'height': img_height
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/fonts/<font_family>', methods=['DELETE'])
|
||||
def delete_font(font_family: str) -> tuple[Response, int] | Response:
|
||||
"""Delete a user-uploaded font file"""
|
||||
try:
|
||||
# Security: Validate font_family to prevent path traversal
|
||||
# Reject if it contains path separators or ..
|
||||
if '..' in font_family or '/' in font_family or '\\' in font_family:
|
||||
return jsonify({'status': 'error', 'message': 'Invalid font family name'}), 400
|
||||
|
||||
# Only allow safe characters: alphanumeric, hyphen, underscore, dot
|
||||
if not re.match(r'^[a-zA-Z0-9_\-\.]+$', font_family):
|
||||
return jsonify({'status': 'error', 'message': 'Invalid font family name'}), 400
|
||||
|
||||
# Check if this is a system font (uses module-level SYSTEM_FONTS frozenset)
|
||||
if font_family.lower() in SYSTEM_FONTS:
|
||||
return jsonify({'status': 'error', 'message': 'Cannot delete system fonts'}), 403
|
||||
|
||||
# Find and delete the font file
|
||||
fonts_dir = PROJECT_ROOT / "assets" / "fonts"
|
||||
|
||||
# Ensure fonts directory exists
|
||||
if not fonts_dir.exists() or not fonts_dir.is_dir():
|
||||
return jsonify({'status': 'error', 'message': 'Fonts directory not found'}), 404
|
||||
|
||||
deleted = False
|
||||
deleted_filename = None
|
||||
|
||||
# Only try valid font extensions (no empty string to avoid matching directories)
|
||||
for ext in ['.ttf', '.otf', '.bdf']:
|
||||
potential_path = fonts_dir / f"{font_family}{ext}"
|
||||
|
||||
# Security: Verify path is within fonts_dir
|
||||
try:
|
||||
potential_path.resolve().relative_to(fonts_dir.resolve())
|
||||
except ValueError:
|
||||
continue # Path escapes fonts_dir, skip
|
||||
|
||||
if potential_path.exists() and potential_path.is_file():
|
||||
potential_path.unlink()
|
||||
deleted = True
|
||||
deleted_filename = f"{font_family}{ext}"
|
||||
break
|
||||
|
||||
if not deleted:
|
||||
# Try case-insensitive match within fonts directory
|
||||
font_family_lower = font_family.lower()
|
||||
for filename in os.listdir(fonts_dir):
|
||||
# Only consider files with valid font extensions
|
||||
if not any(filename.lower().endswith(ext) for ext in ['.ttf', '.otf', '.bdf']):
|
||||
continue
|
||||
|
||||
name_without_ext = os.path.splitext(filename)[0]
|
||||
if name_without_ext.lower() == font_family_lower:
|
||||
filepath = fonts_dir / filename
|
||||
|
||||
# Security: Verify path is within fonts_dir
|
||||
try:
|
||||
filepath.resolve().relative_to(fonts_dir.resolve())
|
||||
except ValueError:
|
||||
continue # Path escapes fonts_dir, skip
|
||||
|
||||
if filepath.is_file():
|
||||
filepath.unlink()
|
||||
deleted = True
|
||||
deleted_filename = filename
|
||||
break
|
||||
|
||||
if not deleted:
|
||||
return jsonify({'status': 'error', 'message': f'Font not found: {font_family}'}), 404
|
||||
|
||||
# Clear font catalog cache
|
||||
try:
|
||||
from web_interface.cache import delete_cached
|
||||
delete_cached('fonts_catalog')
|
||||
except ImportError as e:
|
||||
logger.warning("[FontDelete] Cache module not available: %s", e)
|
||||
except Exception:
|
||||
logger.error("[FontDelete] Failed to clear fonts_catalog cache", exc_info=True)
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': f'Font {deleted_filename} deleted successfully'
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@@ -0,0 +1,398 @@
|
||||
"""Routes with no larger group of their own: errors, integrations,
|
||||
cache, sync, skins, logs, health and hardware.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
ErrorCode, Path, _JOURNALCTL, _SUDO, _get_display_service_status, api_v3,
|
||||
describe_exception, error_response, get_error_aggregator, json, jsonify,
|
||||
logger, os, request, subprocess, success_response,
|
||||
)
|
||||
import web_interface.blueprints.api_v3 as _pkg
|
||||
# Read through the module rather than bound by value: tests patch these
|
||||
# as module attributes, and a value binding would not see the patch.
|
||||
# Several are also called from helpers that live in __init__, so the
|
||||
# package is the only patch point that covers every caller.
|
||||
|
||||
|
||||
@api_v3.route('/health', methods=['GET'])
|
||||
def get_health():
|
||||
"""Get system health status"""
|
||||
try:
|
||||
health_status = {
|
||||
'status': 'healthy',
|
||||
'timestamp': _pkg.time.time(),
|
||||
'services': {},
|
||||
'checks': {}
|
||||
}
|
||||
|
||||
# Check web interface service
|
||||
# Stamp the start _pkg.time before measuring against it -- reading it with a
|
||||
# fallback of _pkg.time.time() and only assigning afterwards made the very
|
||||
# first call subtract two separate clock reads, reporting a small
|
||||
# negative uptime.
|
||||
if not hasattr(get_health, '_start_time'):
|
||||
get_health._start_time = _pkg.time.time()
|
||||
health_status['services']['web_interface'] = {
|
||||
'status': 'running',
|
||||
'uptime_seconds': _pkg.time.time() - get_health._start_time
|
||||
}
|
||||
|
||||
# Check display service
|
||||
display_service_status = _get_display_service_status()
|
||||
health_status['services']['display_service'] = {
|
||||
'status': 'active' if display_service_status.get('active') else 'inactive',
|
||||
'details': display_service_status
|
||||
}
|
||||
|
||||
# Check config file accessibility
|
||||
try:
|
||||
if api_v3.config_manager:
|
||||
test_config = api_v3.config_manager.load_config()
|
||||
health_status['checks']['config_file'] = {
|
||||
'status': 'accessible',
|
||||
'readable': True
|
||||
}
|
||||
else:
|
||||
health_status['checks']['config_file'] = {
|
||||
'status': 'unknown',
|
||||
'readable': False
|
||||
}
|
||||
except Exception as e:
|
||||
health_status['checks']['config_file'] = {
|
||||
'status': 'error',
|
||||
'readable': False,
|
||||
'error': 'see logs for details'
|
||||
}
|
||||
|
||||
# Check plugin system
|
||||
try:
|
||||
if api_v3.plugin_manager:
|
||||
# Try to discover plugins (lightweight check)
|
||||
plugin_count = len(api_v3.plugin_manager.get_available_plugins()) if hasattr(api_v3.plugin_manager, 'get_available_plugins') else 0
|
||||
health_status['checks']['plugin_system'] = {
|
||||
'status': 'operational',
|
||||
'plugin_count': plugin_count
|
||||
}
|
||||
else:
|
||||
health_status['checks']['plugin_system'] = {
|
||||
'status': 'not_initialized'
|
||||
}
|
||||
except Exception as e:
|
||||
health_status['checks']['plugin_system'] = {
|
||||
'status': 'error',
|
||||
'error': 'see logs for details'
|
||||
}
|
||||
|
||||
# Check hardware connectivity (if display manager available)
|
||||
try:
|
||||
snapshot_path = "/tmp/led_matrix_preview.png"
|
||||
if os.path.exists(snapshot_path):
|
||||
# Check if snapshot is recent (updated in last 60 seconds)
|
||||
mtime = os.path.getmtime(snapshot_path)
|
||||
age_seconds = _pkg.time.time() - mtime
|
||||
health_status['checks']['hardware'] = {
|
||||
'status': 'connected' if age_seconds < 60 else 'stale',
|
||||
'snapshot_age_seconds': round(age_seconds, 1)
|
||||
}
|
||||
else:
|
||||
health_status['checks']['hardware'] = {
|
||||
'status': 'no_snapshot',
|
||||
'note': 'Display service may not be running'
|
||||
}
|
||||
except Exception as e:
|
||||
health_status['checks']['hardware'] = {
|
||||
'status': 'unknown',
|
||||
'error': 'see logs for details'
|
||||
}
|
||||
|
||||
# Determine overall health
|
||||
all_healthy = all(
|
||||
check.get('status') in ['accessible', 'operational', 'connected', 'running', 'active']
|
||||
for check in health_status['checks'].values()
|
||||
)
|
||||
|
||||
if not all_healthy:
|
||||
health_status['status'] = 'degraded'
|
||||
|
||||
return jsonify({'status': 'success', 'data': health_status})
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e),
|
||||
'data': {'status': 'unhealthy'}
|
||||
}), 500
|
||||
@api_v3.route('/hardware/status', methods=['GET'])
|
||||
def get_hardware_status():
|
||||
"""Return LED matrix hardware initialization status written by display_manager at startup."""
|
||||
status_path = "/tmp/led_matrix_hw_status.json" # nosec B108
|
||||
try:
|
||||
with open(status_path) as f:
|
||||
hw_data = json.load(f)
|
||||
return jsonify({"status": "success", "data": hw_data})
|
||||
except FileNotFoundError:
|
||||
return jsonify({"status": "success", "data": {"ok": None, "error": "Display service not yet started"}})
|
||||
except PermissionError:
|
||||
logger.warning("Permission denied reading hardware status file; display service may be running as a different user")
|
||||
return jsonify({"status": "success", "data": {"ok": False, "error": "Hardware status temporarily unavailable"}})
|
||||
except json.JSONDecodeError:
|
||||
logger.error("Failed to parse hardware status file", exc_info=True)
|
||||
return jsonify({"status": "success", "data": {"ok": False, "error": "Hardware status file corrupted"}})
|
||||
except Exception:
|
||||
logger.error("Unexpected error reading hardware status", exc_info=True)
|
||||
return jsonify({"status": "error", "message": "Unable to read hardware status"}), 500
|
||||
@api_v3.route('/skins', methods=['GET'])
|
||||
def list_skins():
|
||||
"""List installed visual skins (docs/SKIN_SYSTEM.md).
|
||||
|
||||
Optional ?plugin_id=... filters to skins matching that plugin.
|
||||
"""
|
||||
try:
|
||||
from src.skin_system import skin_runtime
|
||||
|
||||
plugin_id = request.args.get('plugin_id')
|
||||
if plugin_id:
|
||||
skins = skin_runtime.skins_for_plugin(plugin_id)
|
||||
else:
|
||||
# The discovery cache self-invalidates on directory/manifest
|
||||
# mtime changes, so no force_refresh — keeps Pi disk I/O down.
|
||||
skins = skin_runtime.discover_skins()
|
||||
|
||||
payload = []
|
||||
for skin_id, manifest in sorted(skins.items()):
|
||||
skin_dir = Path(manifest['_skin_dir'])
|
||||
preview = manifest.get('preview')
|
||||
payload.append({
|
||||
'id': skin_id,
|
||||
'name': manifest.get('name', skin_id),
|
||||
'version': manifest.get('version'),
|
||||
'author': manifest.get('author'),
|
||||
'description': manifest.get('description', ''),
|
||||
'skin_api_version': manifest.get('skin_api_version'),
|
||||
'targets': manifest.get('targets', {}),
|
||||
'modes': manifest.get('modes', []),
|
||||
'has_preview': bool(preview and (skin_dir / preview).is_file()),
|
||||
})
|
||||
return jsonify({'status': 'success', 'data': {'skins': payload}})
|
||||
except Exception as e:
|
||||
logger.error('Error in list_skins', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/logs', methods=['GET'])
|
||||
def get_logs():
|
||||
"""Get system logs from journalctl"""
|
||||
try:
|
||||
if not _JOURNALCTL:
|
||||
return jsonify({'status': 'error', 'message': 'journalctl not found on this system'}), 503
|
||||
# Get recent logs from journalctl
|
||||
_cmd = ([_SUDO, _JOURNALCTL] if _SUDO else [_JOURNALCTL]) + [
|
||||
'-u', 'ledmatrix.service', '-u', 'ledmatrix-web.service',
|
||||
'-n', '100', '--no-pager', '--output=short-iso']
|
||||
result = subprocess.run(
|
||||
_cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
logs_text = result.stdout.strip()
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'logs': logs_text if logs_text else 'No logs available from ledmatrix or ledmatrix-web service'
|
||||
}
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'Failed to get logs: {result.stderr}'
|
||||
}), 500
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'Timeout while fetching logs'
|
||||
}), 500
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
# Multi-Display Sync Endpoints
|
||||
@api_v3.route('/sync/status', methods=['GET'])
|
||||
def get_sync_status():
|
||||
"""Return live multi-display sync status written by the display process."""
|
||||
import os as _os
|
||||
status_file = "/tmp/led_matrix_sync_status.json"
|
||||
# Also surface config so the UI can show the configured role even before
|
||||
# the display process has written a status file.
|
||||
cfg_role = "standalone"
|
||||
cfg_port = 5765
|
||||
if api_v3.config_manager:
|
||||
try:
|
||||
cfg = api_v3.config_manager.load_config().get("sync", {})
|
||||
cfg_role = cfg.get("role", "standalone")
|
||||
cfg_port = int(cfg.get("port", 5765))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if _os.path.exists(status_file):
|
||||
try:
|
||||
with open(status_file) as f:
|
||||
live = json.load(f)
|
||||
return jsonify({"status": "success", "data": live})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Status file not yet written — return config-only placeholder
|
||||
return jsonify({
|
||||
"status": "success",
|
||||
"data": {
|
||||
"role": cfg_role,
|
||||
"port": cfg_port,
|
||||
"state": "starting",
|
||||
}
|
||||
})
|
||||
@api_v3.route('/cache/list', methods=['GET'])
|
||||
def list_cache_files():
|
||||
"""List all cache files with metadata"""
|
||||
try:
|
||||
if not api_v3.cache_manager:
|
||||
# Initialize cache manager if not already initialized
|
||||
from src.cache_manager import CacheManager
|
||||
api_v3.cache_manager = CacheManager()
|
||||
|
||||
cache_files = api_v3.cache_manager.list_cache_files()
|
||||
cache_dir = api_v3.cache_manager.get_cache_dir()
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'cache_files': cache_files,
|
||||
'cache_dir': cache_dir,
|
||||
'total_files': len(cache_files)
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error('Error in list_cache_files', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/cache/delete', methods=['POST'])
|
||||
def delete_cache_file():
|
||||
"""Delete a specific cache file by key"""
|
||||
try:
|
||||
if not api_v3.cache_manager:
|
||||
# Initialize cache manager if not already initialized
|
||||
from src.cache_manager import CacheManager
|
||||
api_v3.cache_manager = CacheManager()
|
||||
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'key' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
|
||||
|
||||
cache_key = data['key']
|
||||
|
||||
# Delete the cache file
|
||||
api_v3.cache_manager.clear_cache(cache_key)
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': f'Cache file for key "{cache_key}" deleted successfully'
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error('Error in delete_cache_file', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/errors/summary', methods=['GET'])
|
||||
def get_error_summary():
|
||||
"""
|
||||
Get summary of all errors for monitoring and debugging.
|
||||
|
||||
Returns error counts, detected patterns, and recent errors.
|
||||
"""
|
||||
try:
|
||||
aggregator = get_error_aggregator()
|
||||
summary = aggregator.get_error_summary()
|
||||
return success_response(data=summary, message="Error summary retrieved")
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting error summary: {e}", exc_info=True)
|
||||
return error_response(
|
||||
error_code=ErrorCode.SYSTEM_ERROR,
|
||||
message="Failed to retrieve error summary",
|
||||
status_code=500
|
||||
)
|
||||
@api_v3.route('/errors/plugin/<plugin_id>', methods=['GET'])
|
||||
def get_plugin_errors(plugin_id):
|
||||
"""
|
||||
Get error health status for a specific plugin.
|
||||
|
||||
Args:
|
||||
plugin_id: Plugin identifier
|
||||
|
||||
Returns health status and error statistics for the plugin.
|
||||
"""
|
||||
try:
|
||||
aggregator = get_error_aggregator()
|
||||
health = aggregator.get_plugin_health(plugin_id)
|
||||
return success_response(data=health, message="Plugin health retrieved")
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting plugin health for {plugin_id}: {e}", exc_info=True)
|
||||
return error_response(
|
||||
error_code=ErrorCode.SYSTEM_ERROR,
|
||||
message=f"Failed to retrieve health for plugin {plugin_id}",
|
||||
status_code=500
|
||||
)
|
||||
@api_v3.route('/errors/clear', methods=['POST'])
|
||||
def clear_old_errors():
|
||||
"""
|
||||
Clear error records older than specified age.
|
||||
|
||||
Request body (optional):
|
||||
max_age_hours: Maximum age in hours (default: 24, max: 8760 = 1 year)
|
||||
"""
|
||||
try:
|
||||
data = request.get_json(silent=True) or {}
|
||||
raw_max_age = data.get('max_age_hours', 24)
|
||||
|
||||
# Validate and coerce max_age_hours
|
||||
try:
|
||||
max_age_hours = int(raw_max_age)
|
||||
if max_age_hours < 1:
|
||||
return error_response(
|
||||
error_code=ErrorCode.INVALID_INPUT,
|
||||
message="max_age_hours must be at least 1",
|
||||
context={'provided_value': raw_max_age},
|
||||
status_code=400
|
||||
)
|
||||
if max_age_hours > 8760: # 1 year max
|
||||
return error_response(
|
||||
error_code=ErrorCode.INVALID_INPUT,
|
||||
message="max_age_hours cannot exceed 8760 (1 year)",
|
||||
context={'provided_value': raw_max_age},
|
||||
status_code=400
|
||||
)
|
||||
except (ValueError, TypeError, OverflowError):
|
||||
return error_response(
|
||||
error_code=ErrorCode.INVALID_INPUT,
|
||||
message="max_age_hours must be a valid integer",
|
||||
context={'provided_value': str(raw_max_age)},
|
||||
status_code=400
|
||||
)
|
||||
|
||||
aggregator = get_error_aggregator()
|
||||
cleared_count = aggregator.clear_old_records(max_age_hours=max_age_hours)
|
||||
|
||||
return success_response(
|
||||
data={'cleared_count': cleared_count},
|
||||
message=f"Cleared {cleared_count} error records older than {max_age_hours} hours"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error clearing old errors: {e}", exc_info=True)
|
||||
return error_response(
|
||||
error_code=ErrorCode.SYSTEM_ERROR,
|
||||
message="Failed to clear old errors",
|
||||
status_code=500
|
||||
)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,692 @@
|
||||
"""Starlark / Tronbyte app management routes.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
PROJECT_ROOT, Path, _find_pixlet_binary, _install_star_file, _standalone_render_starlark_app,
|
||||
_starlark_github_token, _validate_and_sanitize_app_id,
|
||||
_validate_starlark_app_path, _validate_timing_value, api_v3, describe_exception, json, jsonify,
|
||||
logger, os, request, shutil, subprocess, tempfile,
|
||||
)
|
||||
import web_interface.blueprints.api_v3 as _pkg
|
||||
# Read through the module rather than bound by value: tests patch these
|
||||
# as module attributes, and a value binding would not see the patch.
|
||||
# Several are also called from helpers that live in __init__, so the
|
||||
# package is the only patch point that covers every caller.
|
||||
|
||||
|
||||
@api_v3.route('/starlark/status', methods=['GET'])
|
||||
def get_starlark_status():
|
||||
"""Get Starlark plugin status and Pixlet availability."""
|
||||
try:
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
info = starlark_plugin.get_info()
|
||||
magnify_info = starlark_plugin.get_magnify_recommendation()
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'pixlet_available': info.get('pixlet_available', False),
|
||||
'pixlet_version': info.get('pixlet_version'),
|
||||
'installed_apps': info.get('installed_apps', 0),
|
||||
'enabled_apps': info.get('enabled_apps', 0),
|
||||
'current_app': info.get('current_app'),
|
||||
'plugin_enabled': starlark_plugin.enabled,
|
||||
'display_info': magnify_info
|
||||
})
|
||||
|
||||
# Plugin not loaded - check Pixlet availability via shared resolver
|
||||
# (respects user-configured pixlet_path, bundled binary, and system PATH)
|
||||
full_config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
|
||||
pixlet_path = _find_pixlet_binary(full_config.get('starlark-apps', {}).get('pixlet_path'))
|
||||
pixlet_available = pixlet_path is not None
|
||||
|
||||
# Read app counts from manifest
|
||||
manifest = _pkg._read_starlark_manifest()
|
||||
apps = manifest.get('apps', {})
|
||||
installed_count = len(apps)
|
||||
enabled_count = sum(1 for a in apps.values() if a.get('enabled', True))
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'pixlet_available': pixlet_available,
|
||||
'pixlet_version': None,
|
||||
'installed_apps': installed_count,
|
||||
'enabled_apps': enabled_count,
|
||||
'plugin_enabled': True,
|
||||
'plugin_loaded': False,
|
||||
'display_info': {}
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] get_starlark_status failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to get Starlark status', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/install-pixlet', methods=['POST'])
|
||||
def install_pixlet():
|
||||
"""Download and install Pixlet binary."""
|
||||
try:
|
||||
script_path = PROJECT_ROOT / 'scripts' / 'download_pixlet.sh'
|
||||
if not script_path.exists():
|
||||
return jsonify({'status': 'error', 'message': 'Installation script not found'}), 404
|
||||
|
||||
os.chmod(script_path, 0o755)
|
||||
|
||||
result = subprocess.run(
|
||||
[str(script_path)],
|
||||
cwd=str(PROJECT_ROOT),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=300
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
logger.info("Pixlet downloaded successfully")
|
||||
return jsonify({'status': 'success', 'message': 'Pixlet installed successfully!', 'output': result.stdout})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': f'Failed to download Pixlet: {result.stderr}'}), 500
|
||||
|
||||
except subprocess.TimeoutExpired as err:
|
||||
logger.exception("[Starlark] Pixlet download timed out")
|
||||
return jsonify({'status': 'error', 'message': 'Download timed out',
|
||||
'details': describe_exception(err)}), 500
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] install_pixlet failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to install Pixlet', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/apps', methods=['GET'])
|
||||
def get_starlark_apps():
|
||||
"""List all installed Starlark apps."""
|
||||
try:
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
apps_list = []
|
||||
for app_id, app_instance in starlark_plugin.apps.items():
|
||||
apps_list.append({
|
||||
'id': app_id,
|
||||
'name': app_instance.manifest.get('name', app_id),
|
||||
'enabled': app_instance.is_enabled(),
|
||||
'has_frames': app_instance.frames is not None,
|
||||
'render_interval': app_instance.get_render_interval(),
|
||||
'display_duration': app_instance.get_display_duration(),
|
||||
'config': app_instance.config,
|
||||
'has_schema': app_instance.schema is not None,
|
||||
'last_render_time': app_instance.last_render_time
|
||||
})
|
||||
return jsonify({'status': 'success', 'apps': apps_list, 'count': len(apps_list)})
|
||||
|
||||
# Standalone: read manifest from disk
|
||||
manifest = _pkg._read_starlark_manifest()
|
||||
apps_list = []
|
||||
for app_id, app_data in manifest.get('apps', {}).items():
|
||||
apps_list.append({
|
||||
'id': app_id,
|
||||
'name': app_data.get('name', app_id),
|
||||
'enabled': app_data.get('enabled', True),
|
||||
'has_frames': False,
|
||||
'render_interval': app_data.get('render_interval', 300),
|
||||
'display_duration': app_data.get('display_duration', 15),
|
||||
'config': app_data.get('config', {}),
|
||||
'has_schema': False,
|
||||
'last_render_time': None
|
||||
})
|
||||
return jsonify({'status': 'success', 'apps': apps_list, 'count': len(apps_list)})
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] get_starlark_apps failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to get Starlark apps', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/apps/<app_id>', methods=['GET'])
|
||||
def get_starlark_app(app_id):
|
||||
"""Get details for a specific Starlark app."""
|
||||
try:
|
||||
# Validate app_id before any filesystem access
|
||||
app_dir, error_msg = _validate_starlark_app_path(app_id)
|
||||
if error_msg:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
app = starlark_plugin.apps.get(app_id)
|
||||
if not app:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'app': {
|
||||
'id': app_id,
|
||||
'name': app.manifest.get('name', app_id),
|
||||
'enabled': app.is_enabled(),
|
||||
'config': app.config,
|
||||
'schema': app.schema,
|
||||
'render_interval': app.get_render_interval(),
|
||||
'display_duration': app.get_display_duration(),
|
||||
'has_frames': app.frames is not None,
|
||||
'frame_count': len(app.frames) if app.frames else 0,
|
||||
'last_render_time': app.last_render_time,
|
||||
}
|
||||
})
|
||||
|
||||
# Standalone: read from manifest
|
||||
manifest = _pkg._read_starlark_manifest()
|
||||
app_data = manifest.get('apps', {}).get(app_id)
|
||||
if not app_data:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
|
||||
# Load schema from schema.json if it exists (path already validated above)
|
||||
schema = None
|
||||
schema_file = app_dir / 'schema.json'
|
||||
if schema_file.exists():
|
||||
try:
|
||||
with open(schema_file, 'r') as f:
|
||||
schema = json.load(f)
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
logger.warning(f"Failed to load schema for {app_id}: {e}")
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'app': {
|
||||
'id': app_id,
|
||||
'name': app_data.get('name', app_id),
|
||||
'enabled': app_data.get('enabled', True),
|
||||
'config': app_data.get('config', {}),
|
||||
'schema': schema,
|
||||
'render_interval': app_data.get('render_interval', 300),
|
||||
'display_duration': app_data.get('display_duration', 15),
|
||||
'has_frames': False,
|
||||
'frame_count': 0,
|
||||
'last_render_time': None,
|
||||
}
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] get_starlark_app failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to get Starlark app', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/upload', methods=['POST'])
|
||||
def upload_starlark_app():
|
||||
"""Upload and install a new Starlark app."""
|
||||
try:
|
||||
if 'file' not in request.files:
|
||||
return jsonify({'status': 'error', 'message': 'No file uploaded'}), 400
|
||||
|
||||
file = request.files['file']
|
||||
if not file.filename or not file.filename.endswith('.star'):
|
||||
return jsonify({'status': 'error', 'message': 'File must have .star extension'}), 400
|
||||
|
||||
# Check file size (limit to 5MB for .star files)
|
||||
file.seek(0, 2) # Seek to end
|
||||
file_size = file.tell()
|
||||
file.seek(0) # Reset to beginning
|
||||
MAX_STAR_SIZE = 5 * 1024 * 1024 # 5MB
|
||||
if file_size > MAX_STAR_SIZE:
|
||||
return jsonify({'status': 'error', 'message': f'File too large (max 5MB, got {file_size/1024/1024:.1f}MB)'}), 400
|
||||
|
||||
app_name = request.form.get('name')
|
||||
app_id_input = request.form.get('app_id')
|
||||
filename_base = file.filename.replace('.star', '') if file.filename else None
|
||||
app_id, app_id_error = _validate_and_sanitize_app_id(app_id_input, fallback_source=filename_base)
|
||||
if app_id_error:
|
||||
return jsonify({'status': 'error', 'message': f'Invalid app_id: {app_id_error}'}), 400
|
||||
|
||||
render_interval_input = request.form.get('render_interval')
|
||||
render_interval = 300
|
||||
if render_interval_input is not None:
|
||||
render_interval, err = _validate_timing_value(render_interval_input, 'render_interval')
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
render_interval = render_interval or 300
|
||||
|
||||
display_duration_input = request.form.get('display_duration')
|
||||
display_duration = 15
|
||||
if display_duration_input is not None:
|
||||
display_duration, err = _validate_timing_value(display_duration_input, 'display_duration')
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
display_duration = display_duration or 15
|
||||
|
||||
import tempfile
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.star') as tmp:
|
||||
file.save(tmp.name)
|
||||
temp_path = tmp.name
|
||||
|
||||
try:
|
||||
metadata = {'name': app_name or app_id, 'render_interval': render_interval, 'display_duration': display_duration}
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
success = starlark_plugin.install_app(app_id, temp_path, metadata)
|
||||
else:
|
||||
success = _install_star_file(app_id, temp_path, metadata)
|
||||
if success:
|
||||
return jsonify({'status': 'success', 'message': f'App installed: {app_id}', 'app_id': app_id})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Failed to install app'}), 500
|
||||
finally:
|
||||
try:
|
||||
os.unlink(temp_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
except (OSError, IOError) as err:
|
||||
# This used to withhold the detail because it names absolute paths on
|
||||
# the device. A full disk and a bad permission are indistinguishable
|
||||
# without it, though, and describe_exception redacts credentials and
|
||||
# truncates -- the same trade-off every other handler here makes.
|
||||
logger.exception("[Starlark] File error uploading starlark app: %s", err)
|
||||
return jsonify({'status': 'error', 'message': 'File error during upload',
|
||||
'details': describe_exception(err)}), 500
|
||||
except ImportError as err:
|
||||
logger.exception("[Starlark] Module load error uploading starlark app: %s", err)
|
||||
return jsonify({'status': 'error', 'message': 'Failed to load app module', 'details': describe_exception(err)}), 500
|
||||
except Exception as err:
|
||||
logger.exception("[Starlark] Unexpected error uploading starlark app: %s", err)
|
||||
return jsonify({'status': 'error', 'message': 'Failed to upload app', 'details': describe_exception(err)}), 500
|
||||
@api_v3.route('/starlark/apps/<app_id>', methods=['DELETE'])
|
||||
def uninstall_starlark_app(app_id):
|
||||
"""Uninstall a Starlark app."""
|
||||
try:
|
||||
# Validate app_id before any filesystem access
|
||||
app_dir, error_msg = _validate_starlark_app_path(app_id)
|
||||
if error_msg:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
success = starlark_plugin.uninstall_app(app_id)
|
||||
else:
|
||||
# Standalone: remove app dir and manifest entry. app_dir is the
|
||||
# path _validate_starlark_app_path checked, not a fresh join.
|
||||
import shutil
|
||||
if app_dir.exists():
|
||||
shutil.rmtree(app_dir)
|
||||
manifest = _pkg._read_starlark_manifest()
|
||||
manifest.get('apps', {}).pop(app_id, None)
|
||||
success = _pkg._write_starlark_manifest(manifest)
|
||||
|
||||
if success:
|
||||
return jsonify({'status': 'success', 'message': f'App uninstalled: {app_id}'})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Failed to uninstall app'}), 500
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] uninstall_starlark_app failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to uninstall Starlark app', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/apps/<app_id>/config', methods=['GET'])
|
||||
def get_starlark_app_config(app_id):
|
||||
"""Get configuration for a Starlark app."""
|
||||
try:
|
||||
# Validate app_id before any filesystem access
|
||||
app_dir, error_msg = _validate_starlark_app_path(app_id)
|
||||
if error_msg:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
app = starlark_plugin.apps.get(app_id)
|
||||
if not app:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
return jsonify({'status': 'success', 'config': app.config, 'schema': app.schema})
|
||||
|
||||
# Standalone: read from config.json. app_dir is the path
|
||||
# _validate_starlark_app_path checked, not a fresh join.
|
||||
config_file = app_dir / "config.json"
|
||||
|
||||
if not app_dir.exists():
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
|
||||
config = {}
|
||||
if config_file.exists():
|
||||
try:
|
||||
with open(config_file, 'r') as f:
|
||||
config = json.load(f)
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
logger.warning(f"Failed to load config for {app_id}: {e}")
|
||||
|
||||
# Load schema from schema.json
|
||||
schema = None
|
||||
schema_file = app_dir / "schema.json"
|
||||
if schema_file.exists():
|
||||
try:
|
||||
with open(schema_file, 'r') as f:
|
||||
schema = json.load(f)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to load schema for {app_id}: {e}")
|
||||
|
||||
return jsonify({'status': 'success', 'config': config, 'schema': schema})
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] get_starlark_app_config failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to get Starlark app config', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/apps/<app_id>/config', methods=['PUT'])
|
||||
def update_starlark_app_config(app_id):
|
||||
"""Update configuration for a Starlark app."""
|
||||
try:
|
||||
# Validate app_id before any filesystem access
|
||||
app_dir, error_msg = _validate_starlark_app_path(app_id)
|
||||
if error_msg:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No configuration provided'}), 400
|
||||
|
||||
if 'render_interval' in data:
|
||||
val, err = _validate_timing_value(data['render_interval'], 'render_interval')
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
data['render_interval'] = val
|
||||
|
||||
if 'display_duration' in data:
|
||||
val, err = _validate_timing_value(data['display_duration'], 'display_duration')
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
data['display_duration'] = val
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
app = starlark_plugin.apps.get(app_id)
|
||||
if not app:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
|
||||
# Extract timing keys from data before updating config (they belong in manifest, not config)
|
||||
render_interval = data.pop('render_interval', None)
|
||||
display_duration = data.pop('display_duration', None)
|
||||
|
||||
# Snapshot before mutating. save_config() can fail, and the route
|
||||
# answers 500 below when it does -- but the loaded app kept the new
|
||||
# values anyway, so a later GET returned configuration that was
|
||||
# never persisted and the plugin rendered with it.
|
||||
prev_config = dict(app.config)
|
||||
prev_manifest = dict(app.manifest)
|
||||
|
||||
# Update config with non-timing fields only
|
||||
app.config.update(data)
|
||||
|
||||
# Update manifest with timing fields
|
||||
timing_changed = False
|
||||
if render_interval is not None:
|
||||
app.manifest['render_interval'] = render_interval
|
||||
timing_changed = True
|
||||
if display_duration is not None:
|
||||
app.manifest['display_duration'] = display_duration
|
||||
timing_changed = True
|
||||
saved = app.save_config()
|
||||
if not saved:
|
||||
app.config.clear(); app.config.update(prev_config)
|
||||
app.manifest.clear(); app.manifest.update(prev_manifest)
|
||||
if saved:
|
||||
# Persist manifest if timing changed (same pattern as toggle endpoint)
|
||||
if timing_changed:
|
||||
try:
|
||||
# Use safe manifest update to prevent race conditions
|
||||
timing_updates = {}
|
||||
if render_interval is not None:
|
||||
timing_updates['render_interval'] = render_interval
|
||||
if display_duration is not None:
|
||||
timing_updates['display_duration'] = display_duration
|
||||
|
||||
def update_fn(manifest):
|
||||
manifest['apps'][app_id].update(timing_updates)
|
||||
starlark_plugin._update_manifest_safe(update_fn)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to persist timing to manifest for {app_id}: {e}")
|
||||
starlark_plugin._render_app(app, force=True)
|
||||
return jsonify({'status': 'success', 'message': 'Configuration updated', 'config': app.config})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Failed to save configuration'}), 500
|
||||
|
||||
# Standalone: update both config.json and manifest
|
||||
manifest = _pkg._read_starlark_manifest()
|
||||
app_data = manifest.get('apps', {}).get(app_id)
|
||||
if not app_data:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
|
||||
# Extract timing keys (they go in manifest, not config.json)
|
||||
render_interval = data.pop('render_interval', None)
|
||||
display_duration = data.pop('display_duration', None)
|
||||
|
||||
# Update manifest with timing values
|
||||
if render_interval is not None:
|
||||
app_data['render_interval'] = render_interval
|
||||
if display_duration is not None:
|
||||
app_data['display_duration'] = display_duration
|
||||
|
||||
# Load current config from config.json. app_dir is the path
|
||||
# _validate_starlark_app_path checked, not a fresh join.
|
||||
config_file = app_dir / "config.json"
|
||||
current_config = {}
|
||||
if config_file.exists():
|
||||
try:
|
||||
with open(config_file, 'r') as f:
|
||||
current_config = json.load(f)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to load config for {app_id}: {e}")
|
||||
|
||||
# Update config with new values (excluding timing keys)
|
||||
current_config.update(data)
|
||||
|
||||
# Write updated config to config.json
|
||||
try:
|
||||
with open(config_file, 'w') as f:
|
||||
json.dump(current_config, f, indent=2)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to save config.json for {app_id}: {e}")
|
||||
logger.exception("Failed to save Starlark configuration for %r", app_id)
|
||||
return jsonify({'status': 'error', 'message': 'Failed to save configuration',
|
||||
'details': describe_exception(e)}), 500
|
||||
|
||||
# Also update manifest for backward compatibility
|
||||
app_data.setdefault('config', {}).update(data)
|
||||
|
||||
if _pkg._write_starlark_manifest(manifest):
|
||||
return jsonify({'status': 'success', 'message': 'Configuration updated', 'config': current_config})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Failed to save manifest'}), 500
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] update_starlark_app_config failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to update Starlark app config', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/apps/<app_id>/toggle', methods=['POST'])
|
||||
def toggle_starlark_app(app_id):
|
||||
"""Enable or disable a Starlark app."""
|
||||
try:
|
||||
data = request.get_json(silent=True) or {}
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
app = starlark_plugin.apps.get(app_id)
|
||||
if not app:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
enabled = data.get('enabled')
|
||||
if enabled is None:
|
||||
enabled = not app.is_enabled()
|
||||
app.manifest['enabled'] = enabled
|
||||
# Use safe manifest update to prevent race conditions
|
||||
def update_fn(manifest):
|
||||
manifest['apps'][app_id]['enabled'] = enabled
|
||||
starlark_plugin._update_manifest_safe(update_fn)
|
||||
return jsonify({'status': 'success', 'message': f"App {'enabled' if enabled else 'disabled'}", 'enabled': enabled})
|
||||
|
||||
# Standalone: update manifest directly
|
||||
manifest = _pkg._read_starlark_manifest()
|
||||
app_data = manifest.get('apps', {}).get(app_id)
|
||||
if not app_data:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
|
||||
enabled = data.get('enabled')
|
||||
if enabled is None:
|
||||
enabled = not app_data.get('enabled', True)
|
||||
app_data['enabled'] = enabled
|
||||
if _pkg._write_starlark_manifest(manifest):
|
||||
return jsonify({'status': 'success', 'message': f"App {'enabled' if enabled else 'disabled'}", 'enabled': enabled})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Failed to save'}), 500
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] toggle_starlark_app failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to toggle Starlark app', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/apps/<app_id>/render', methods=['POST'])
|
||||
def render_starlark_app(app_id):
|
||||
"""Force render a Starlark app."""
|
||||
try:
|
||||
app_dir, err = _validate_starlark_app_path(app_id)
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
app = starlark_plugin.apps.get(app_id)
|
||||
if not app:
|
||||
return jsonify({'status': 'error', 'message': f'App not found: {app_id}'}), 404
|
||||
success = starlark_plugin._render_app(app, force=True)
|
||||
if success:
|
||||
return jsonify({'status': 'success', 'message': 'App rendered',
|
||||
'frame_count': len(app.frames) if app.frames else 0})
|
||||
return jsonify({'status': 'error', 'message': 'Failed to render app'}), 500
|
||||
|
||||
# Web-service context: plugin not loaded, call pixlet directly
|
||||
success, status_code, error = _standalone_render_starlark_app(app_id)
|
||||
if success:
|
||||
return jsonify({'status': 'success', 'message': 'App rendered successfully', 'frame_count': 0}), status_code
|
||||
return jsonify({'status': 'error', 'message': error or 'Render failed', 'frame_count': 0}), status_code
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] render_starlark_app failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to render Starlark app', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/repository/browse', methods=['GET'])
|
||||
def browse_tronbyte_repository():
|
||||
"""Browse all apps in the Tronbyte repository (bulk cached fetch).
|
||||
|
||||
Returns ALL apps with metadata, categories, and authors.
|
||||
Filtering/sorting/pagination is handled client-side.
|
||||
Results are cached server-side for 2 hours.
|
||||
"""
|
||||
try:
|
||||
TronbyteRepository = _pkg._get_tronbyte_repository_class()
|
||||
|
||||
repo = TronbyteRepository(github_token=_starlark_github_token())
|
||||
|
||||
result = repo.list_all_apps_cached()
|
||||
|
||||
rate_limit = repo.get_rate_limit_info()
|
||||
|
||||
# An upstream failure used to arrive here as an empty app list and go
|
||||
# out as 'success', so the store drew an empty grid and said nothing.
|
||||
# 502: the request was fine, GitHub was not.
|
||||
if result.get('error'):
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': result['error'],
|
||||
'rate_limit': rate_limit,
|
||||
}), 502
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'apps': result['apps'],
|
||||
'categories': result['categories'],
|
||||
'authors': result['authors'],
|
||||
'count': result['count'],
|
||||
'cached': result['cached'],
|
||||
'rate_limit': rate_limit,
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] browse_tronbyte_repository failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to browse repository', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/repository/install', methods=['POST'])
|
||||
def install_from_tronbyte_repository():
|
||||
"""Install an app from the Tronbyte repository."""
|
||||
try:
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'app_id' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'app_id is required'}), 400
|
||||
|
||||
app_id, app_id_error = _validate_and_sanitize_app_id(data['app_id'])
|
||||
if app_id_error:
|
||||
return jsonify({'status': 'error', 'message': f'Invalid app_id: {app_id_error}'}), 400
|
||||
|
||||
TronbyteRepository = _pkg._get_tronbyte_repository_class()
|
||||
import tempfile
|
||||
|
||||
repo = TronbyteRepository(github_token=_starlark_github_token())
|
||||
|
||||
success, metadata, error = repo.get_app_metadata(data['app_id'])
|
||||
if not success:
|
||||
return jsonify({'status': 'error', 'message': f'Failed to fetch app metadata: {error}'}), 404
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.star') as tmp:
|
||||
temp_path = tmp.name
|
||||
|
||||
try:
|
||||
# Pass filename from metadata (e.g., "analog_clock.star" for analogclock app)
|
||||
# Note: manifest uses 'fileName' (camelCase), not 'filename'
|
||||
filename = metadata.get('fileName') if metadata else None
|
||||
success, error = repo.download_star_file(data['app_id'], Path(temp_path), filename=filename)
|
||||
if not success:
|
||||
return jsonify({'status': 'error', 'message': f'Failed to download app: {error}'}), 500
|
||||
|
||||
# Download assets (images, sources, etc.) to a temp directory
|
||||
import tempfile
|
||||
temp_assets_dir = tempfile.mkdtemp()
|
||||
try:
|
||||
success_assets, error_assets = repo.download_app_assets(data['app_id'], Path(temp_assets_dir))
|
||||
# Asset download is non-critical - log warning but continue if it fails
|
||||
if not success_assets:
|
||||
logger.warning(f"Failed to download assets for {data['app_id']}: {error_assets}")
|
||||
|
||||
render_interval = data.get('render_interval', 300)
|
||||
ri, err = _validate_timing_value(render_interval, 'render_interval')
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
render_interval = ri or 300
|
||||
|
||||
display_duration = data.get('display_duration', 15)
|
||||
dd, err = _validate_timing_value(display_duration, 'display_duration')
|
||||
if err:
|
||||
return jsonify({'status': 'error', 'message': err}), 400
|
||||
display_duration = dd or 15
|
||||
|
||||
install_metadata = {
|
||||
'name': metadata.get('name', app_id) if metadata else app_id,
|
||||
'render_interval': render_interval,
|
||||
'display_duration': display_duration
|
||||
}
|
||||
|
||||
starlark_plugin = _pkg._get_starlark_plugin()
|
||||
if starlark_plugin:
|
||||
success = starlark_plugin.install_app(app_id, temp_path, install_metadata, assets_dir=temp_assets_dir)
|
||||
else:
|
||||
success = _install_star_file(app_id, temp_path, install_metadata, assets_dir=temp_assets_dir)
|
||||
finally:
|
||||
# Clean up temp assets directory
|
||||
import shutil
|
||||
try:
|
||||
shutil.rmtree(temp_assets_dir)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
if success:
|
||||
return jsonify({'status': 'success', 'message': f'App installed: {metadata.get("name", app_id) if metadata else app_id}', 'app_id': app_id})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Failed to install app'}), 500
|
||||
finally:
|
||||
try:
|
||||
os.unlink(temp_path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] install_from_tronbyte_repository failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to install from repository', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/starlark/repository/categories', methods=['GET'])
|
||||
def get_tronbyte_categories():
|
||||
"""Get list of available app categories (uses bulk cache)."""
|
||||
try:
|
||||
TronbyteRepository = _pkg._get_tronbyte_repository_class()
|
||||
repo = TronbyteRepository(github_token=_starlark_github_token())
|
||||
|
||||
result = repo.list_all_apps_cached()
|
||||
|
||||
if result.get('error'):
|
||||
return jsonify({'status': 'error', 'message': result['error']}), 502
|
||||
|
||||
return jsonify({'status': 'success', 'categories': result['categories']})
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("[Starlark] get_tronbyte_categories failed")
|
||||
return jsonify({'status': 'error', 'message': 'Failed to fetch categories', 'details': describe_exception(e)}), 500
|
||||
@@ -0,0 +1,619 @@
|
||||
"""Service control, updates, versions and system status.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
Any, Dict, PROJECT_ROOT, Path, _GIT, _UPDATE_CHECK_TTL,
|
||||
_describe_git_failure, _get_display_service_status, _git_current_branch,
|
||||
_git_remote_branch_exists, _git_upstream, _pip_install_requirements,
|
||||
_scrub_git_remote_url, _truncate_output, _update_check_cache,
|
||||
_update_check_failed, api_v3, checkout_branch, describe_exception,
|
||||
get_git_version, jsonify, logger, os, request, resolve_pull_command,
|
||||
shutil, subprocess,
|
||||
)
|
||||
import web_interface.blueprints.api_v3 as _pkg
|
||||
# Read through the module rather than bound by value: tests patch these
|
||||
# as module attributes, and a value binding would not see the patch.
|
||||
# Several are also called from helpers that live in __init__, so the
|
||||
# package is the only patch point that covers every caller.
|
||||
|
||||
|
||||
@api_v3.route('/system/status', methods=['GET'])
|
||||
def get_system_status():
|
||||
"""Get system status"""
|
||||
try:
|
||||
# Check cache first (10 second TTL for system status)
|
||||
try:
|
||||
from web_interface.cache import get_cached, set_cached
|
||||
cached_result = get_cached('system_status', ttl_seconds=10)
|
||||
if cached_result is not None:
|
||||
return jsonify({'status': 'success', 'data': cached_result})
|
||||
except ImportError:
|
||||
# Cache not available, continue without caching
|
||||
get_cached = None
|
||||
set_cached = None
|
||||
|
||||
# Import psutil for system monitoring
|
||||
try:
|
||||
import psutil
|
||||
except ImportError:
|
||||
# Fallback if psutil not available
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'psutil not available for system monitoring'
|
||||
}), 503
|
||||
|
||||
# Get system metrics using psutil
|
||||
cpu_percent = psutil.cpu_percent(interval=0.1) # Short interval for responsiveness
|
||||
memory = psutil.virtual_memory()
|
||||
memory_percent = memory.percent
|
||||
disk = psutil.disk_usage('/')
|
||||
disk_percent = disk.percent
|
||||
|
||||
# Calculate uptime
|
||||
boot_time = psutil.boot_time()
|
||||
uptime_seconds = _pkg.time.time() - boot_time
|
||||
uptime_hours = uptime_seconds / 3600
|
||||
uptime_days = uptime_hours / 24
|
||||
|
||||
# Format uptime string
|
||||
if uptime_days >= 1:
|
||||
uptime_str = f"{int(uptime_days)}d {int(uptime_hours % 24)}h"
|
||||
elif uptime_hours >= 1:
|
||||
uptime_str = f"{int(uptime_hours)}h {int((uptime_seconds % 3600) / 60)}m"
|
||||
else:
|
||||
uptime_str = f"{int(uptime_seconds / 60)}m"
|
||||
|
||||
# Get CPU temperature (Raspberry Pi)
|
||||
cpu_temp = None
|
||||
try:
|
||||
temp_file = '/sys/class/thermal/thermal_zone0/temp'
|
||||
if os.path.exists(temp_file):
|
||||
with open(temp_file, 'r') as f:
|
||||
temp_millidegrees = int(f.read().strip())
|
||||
cpu_temp = temp_millidegrees / 1000.0 # Convert to Celsius
|
||||
except (IOError, ValueError, OSError):
|
||||
# Temperature sensor not available or error reading
|
||||
cpu_temp = None
|
||||
|
||||
# Get display service status
|
||||
service_status = _get_display_service_status()
|
||||
|
||||
status = {
|
||||
'timestamp': _pkg.time.time(),
|
||||
'uptime': uptime_str,
|
||||
'uptime_seconds': int(uptime_seconds),
|
||||
'service_active': service_status.get('active', False),
|
||||
'cpu_percent': round(cpu_percent, 1),
|
||||
'memory_used_percent': round(memory_percent, 1),
|
||||
'memory_total_mb': round(memory.total / (1024 * 1024), 1),
|
||||
'memory_used_mb': round(memory.used / (1024 * 1024), 1),
|
||||
# MemAvailable, not total-minus-used: it accounts for reclaimable
|
||||
# page cache, so it is what actually predicts memory trouble. A
|
||||
# board can read 70% "used" and be fine, or read the same and be
|
||||
# about to fail fork(), and only this number tells them apart.
|
||||
'memory_available_mb': round(memory.available / (1024 * 1024), 1),
|
||||
'cpu_temp': round(cpu_temp, 1) if cpu_temp is not None else None,
|
||||
'disk_used_percent': round(disk_percent, 1),
|
||||
'disk_total_gb': round(disk.total / (1024 * 1024 * 1024), 1),
|
||||
'disk_used_gb': round(disk.used / (1024 * 1024 * 1024), 1)
|
||||
}
|
||||
|
||||
# Cache the result if available
|
||||
if set_cached:
|
||||
try:
|
||||
set_cached('system_status', status, ttl_seconds=10)
|
||||
except Exception:
|
||||
pass # Cache write failed, but continue
|
||||
|
||||
return jsonify({'status': 'success', 'data': status})
|
||||
except Exception as e:
|
||||
logger.error('Unhandled exception', exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
|
||||
@api_v3.route('/system/version', methods=['GET'])
|
||||
def get_system_version():
|
||||
"""Get LEDMatrix repository version"""
|
||||
try:
|
||||
version = get_git_version()
|
||||
return jsonify({'status': 'success', 'data': {'version': version}})
|
||||
except Exception as e:
|
||||
logger.error("get_system_version failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'Unable to retrieve version'}), 500
|
||||
@api_v3.route('/system/check-update', methods=['GET'])
|
||||
def check_for_update():
|
||||
"""Check whether a newer LEDMatrix commit is available on origin/main."""
|
||||
now = _pkg.time.time()
|
||||
if _update_check_cache['result'] and now - _update_check_cache['ts'] < _UPDATE_CHECK_TTL:
|
||||
return jsonify(_update_check_cache['result'])
|
||||
|
||||
_safe: Dict[str, Any] = {'update_available': False, 'remote_sha': 'unknown', 'commits_behind': 0}
|
||||
try:
|
||||
cwd = str(PROJECT_ROOT)
|
||||
fetch_result = subprocess.run(
|
||||
['git', 'fetch', 'origin', 'main', '--quiet'],
|
||||
capture_output=True, timeout=10, cwd=cwd,
|
||||
)
|
||||
if fetch_result.returncode != 0:
|
||||
stderr = fetch_result.stderr.decode(errors='replace').strip()
|
||||
logger.warning("check-update: git fetch failed (rc=%d): %s",
|
||||
fetch_result.returncode, stderr)
|
||||
failed = _update_check_failed(_describe_git_failure(stderr))
|
||||
_update_check_cache['result'] = failed
|
||||
_update_check_cache['ts'] = now
|
||||
return jsonify(failed)
|
||||
local = subprocess.run(
|
||||
['git', 'rev-parse', 'HEAD'],
|
||||
capture_output=True, text=True, timeout=5, cwd=cwd,
|
||||
).stdout.strip()
|
||||
remote = subprocess.run(
|
||||
['git', 'rev-parse', 'origin/main'],
|
||||
capture_output=True, text=True, timeout=5, cwd=cwd,
|
||||
).stdout.strip()
|
||||
|
||||
if not local or not remote:
|
||||
return jsonify(_safe)
|
||||
|
||||
if local == remote:
|
||||
result: Dict[str, Any] = {'update_available': False, 'remote_sha': remote, 'commits_behind': 0}
|
||||
else:
|
||||
count_str = subprocess.run(
|
||||
['git', 'rev-list', 'HEAD..origin/main', '--count'],
|
||||
capture_output=True, text=True, timeout=5, cwd=cwd,
|
||||
).stdout.strip()
|
||||
count = int(count_str) if count_str.isdigit() else 0
|
||||
result = {'update_available': count > 0, 'remote_sha': remote, 'commits_behind': count}
|
||||
|
||||
_update_check_cache['result'] = result
|
||||
_update_check_cache['ts'] = now
|
||||
return jsonify(result)
|
||||
except Exception as e:
|
||||
logger.warning("check-update failed: %s", e)
|
||||
return jsonify(_update_check_failed(
|
||||
"Could not check for updates; see logs for details."))
|
||||
@api_v3.route('/system/action', methods=['POST'])
|
||||
def execute_system_action():
|
||||
"""Execute system actions (start/stop/reboot/etc)"""
|
||||
try:
|
||||
# HTMX sends data as form data, not JSON
|
||||
data = request.get_json(silent=True) or {}
|
||||
if not data:
|
||||
# Try to get from form data if JSON fails
|
||||
data = {
|
||||
'action': request.form.get('action'),
|
||||
'mode': request.form.get('mode')
|
||||
}
|
||||
|
||||
if not data or 'action' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'Action required'}), 400
|
||||
|
||||
action = data['action']
|
||||
mode = data.get('mode') # For on-demand modes
|
||||
|
||||
# Map actions to subprocess calls (similar to original implementation)
|
||||
if action == 'start_display':
|
||||
if mode:
|
||||
# For on-demand modes, we would need to integrate with the display controller
|
||||
# For now, just start the display service
|
||||
try:
|
||||
result = subprocess.run(['sudo', 'systemctl', 'start', 'ledmatrix.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
except subprocess.TimeoutExpired as e:
|
||||
logger.error("start_display (%s) timed out: %s", mode, e)
|
||||
return jsonify({'status': 'error', 'message': 'Command timed out', 'returncode': -1, 'stderr': 'timeout'})
|
||||
logger.info("start_display (%s) returned code %d", mode, result.returncode)
|
||||
if result.returncode != 0 and result.stderr:
|
||||
logger.error("start_display (%s) stderr: %s", mode, result.stderr.strip())
|
||||
resp = {
|
||||
'status': 'success' if result.returncode == 0 else 'error',
|
||||
'message': 'Display started' if result.returncode == 0 else 'Failed to start display',
|
||||
}
|
||||
if result.returncode != 0:
|
||||
resp['returncode'] = result.returncode
|
||||
resp['stderr'] = result.stderr.strip()
|
||||
return jsonify(resp)
|
||||
else:
|
||||
result = subprocess.run(['sudo', 'systemctl', 'start', 'ledmatrix.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'stop_display':
|
||||
result = subprocess.run(['sudo', 'systemctl', 'stop', 'ledmatrix.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'enable_autostart':
|
||||
result = subprocess.run(['sudo', 'systemctl', 'enable', 'ledmatrix.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'disable_autostart':
|
||||
result = subprocess.run(['sudo', 'systemctl', 'disable', 'ledmatrix.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'reboot_system':
|
||||
result = subprocess.run(['sudo', 'reboot'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'shutdown_system':
|
||||
result = subprocess.run(['sudo', 'poweroff'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'git_pull':
|
||||
# Use PROJECT_ROOT instead of hardcoded path
|
||||
project_dir = str(PROJECT_ROOT)
|
||||
|
||||
# Decide how to pull BEFORE stashing. If this checkout cannot be
|
||||
# updated at all, stashing first would put the user's local changes
|
||||
# away for an update that was never going to run.
|
||||
pull_args, upstream_note, pull_error = resolve_pull_command(project_dir)
|
||||
if pull_error:
|
||||
logger.warning("git pull not attempted: %s", pull_error)
|
||||
return jsonify({'status': 'error', 'message': pull_error})
|
||||
|
||||
# Check if there are local changes that need to be stashed
|
||||
# Exclude plugins directory - plugins are separate repos and shouldn't be stashed with base project
|
||||
# Use --untracked-files=no to skip untracked files check (much faster with symlinked plugins)
|
||||
try:
|
||||
status_result = subprocess.run(
|
||||
['git', 'status', '--porcelain', '--untracked-files=no'],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
cwd=project_dir
|
||||
)
|
||||
# Filter out any changes in plugins directory - plugins are separate repositories
|
||||
# Git status format: XY filename (where X is status of index, Y is status of work tree)
|
||||
status_lines = [line for line in status_result.stdout.strip().split('\n')
|
||||
if line.strip() and 'plugins/' not in line]
|
||||
has_changes = bool('\n'.join(status_lines).strip())
|
||||
except subprocess.TimeoutExpired:
|
||||
# If status check times out, assume there might be changes and proceed
|
||||
# This is safer than failing the update
|
||||
has_changes = True
|
||||
status_result = type('obj', (object,), {'stdout': '', 'stderr': 'Status check timed out'})()
|
||||
|
||||
stash_info = ""
|
||||
|
||||
# Stash local changes if they exist (excluding plugins)
|
||||
# Plugins are separate repositories and shouldn't be stashed with base project updates
|
||||
if has_changes:
|
||||
try:
|
||||
# Use pathspec to exclude plugins directory from stash
|
||||
stash_result = subprocess.run(
|
||||
['git', 'stash', 'push', '-m', 'LEDMatrix auto-stash before update', '--', ':!plugins'],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
cwd=project_dir
|
||||
)
|
||||
if stash_result.returncode == 0:
|
||||
logger.debug("git stash: stashed local changes before pull")
|
||||
stash_info = " Local changes were stashed."
|
||||
else:
|
||||
logger.warning("git stash failed before pull (returncode=%d)", stash_result.returncode)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("git stash timed out, proceeding with pull")
|
||||
|
||||
# Record HEAD before the pull so dependency changes can be detected
|
||||
old_head = None
|
||||
try:
|
||||
_pre = subprocess.run(['git', 'rev-parse', 'HEAD'],
|
||||
capture_output=True, text=True, timeout=10, cwd=project_dir)
|
||||
if _pre.returncode == 0:
|
||||
old_head = _pre.stdout.strip()
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("git rev-parse timed out before pull")
|
||||
|
||||
# Whether the pull actually brought new code in. "Already up to
|
||||
# date" is a success too, and prompting for a restart then would
|
||||
# train users to ignore the prompt.
|
||||
code_changed = False
|
||||
|
||||
# Perform the git pull. Branches without an upstream were given
|
||||
# an explicit "origin <branch>" above so the update still works.
|
||||
result = subprocess.run(
|
||||
pull_args,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
cwd=project_dir
|
||||
)
|
||||
|
||||
# Give the branch tracking information so the next pull is a plain
|
||||
# `git pull` — otherwise every update repeats the fallback.
|
||||
if result.returncode == 0 and upstream_note:
|
||||
branch = _git_current_branch(project_dir)
|
||||
if branch:
|
||||
try:
|
||||
subprocess.run(
|
||||
['git', 'branch', f'--set-upstream-to=origin/{branch}', branch],
|
||||
capture_output=True, text=True, timeout=10, cwd=project_dir)
|
||||
except (subprocess.TimeoutExpired, OSError) as exc:
|
||||
logger.debug("could not set upstream for %s: %s", branch, exc)
|
||||
|
||||
# Return custom response for git_pull
|
||||
if result.returncode == 0:
|
||||
pull_message = "Code updated successfully."
|
||||
if has_changes:
|
||||
pull_message = f"Code updated successfully. Local changes were automatically stashed.{stash_info}"
|
||||
if result.stdout and "Already up to date" not in result.stdout:
|
||||
pull_message = f"Code updated successfully.{stash_info}"
|
||||
if upstream_note:
|
||||
pull_message = f"{pull_message} {upstream_note}"
|
||||
|
||||
# Keep Python dependencies in sync automatically: if the pull
|
||||
# changed a requirements file, install it now — users updating
|
||||
# from the web UI (most of them) never SSH in to pip install.
|
||||
# Installs go through the same root-visible path as the
|
||||
# Tools-tab buttons (_pip_install_requirements).
|
||||
dep_notes = []
|
||||
try:
|
||||
_post = subprocess.run(['git', 'rev-parse', 'HEAD'],
|
||||
capture_output=True, text=True, timeout=10, cwd=project_dir)
|
||||
new_head = _post.stdout.strip() if _post.returncode == 0 else None
|
||||
if old_head and new_head and old_head != new_head:
|
||||
code_changed = True
|
||||
diff = subprocess.run(
|
||||
['git', 'diff', '--name-only', f'{old_head}..{new_head}'],
|
||||
capture_output=True, text=True, timeout=15, cwd=project_dir)
|
||||
changed = set(diff.stdout.split()) if diff.returncode == 0 else set()
|
||||
for rel in ('requirements.txt', 'web_interface/requirements.txt'):
|
||||
req_path = PROJECT_ROOT / rel
|
||||
if rel not in changed or not req_path.exists():
|
||||
continue
|
||||
# Each file's install is isolated: a timeout or
|
||||
# OSError (e.g. the sudo wrapper/interpreter
|
||||
# missing) on one file must not abort the other.
|
||||
try:
|
||||
r = _pip_install_requirements(req_path, timeout=180)
|
||||
if r.returncode == 0:
|
||||
dep_notes.append(f"Dependencies from {rel} updated.")
|
||||
else:
|
||||
dep_notes.append(
|
||||
f"Dependency install from {rel} failed — "
|
||||
"run Install Base Requirements from the Tools tab.")
|
||||
logger.warning("post-update pip install failed for %s: %s",
|
||||
rel, _truncate_output(r.stdout, r.stderr))
|
||||
except subprocess.TimeoutExpired:
|
||||
dep_notes.append(
|
||||
f"Dependency install from {rel} timed out — "
|
||||
"run Install Base Requirements from the Tools tab.")
|
||||
logger.warning("post-update pip install timed out for %s", rel)
|
||||
except OSError as install_err:
|
||||
dep_notes.append(
|
||||
f"Dependency install from {rel} failed — "
|
||||
"run Install Base Requirements from the Tools tab.")
|
||||
logger.warning("post-update pip install errored for %s: %s",
|
||||
rel, install_err)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("post-update dependency sync timed out")
|
||||
if dep_notes:
|
||||
pull_message += " " + " ".join(dep_notes)
|
||||
# A `git pull` restores built-in plugins (committed under
|
||||
# plugin-repos/) even if the user uninstalled them. Re-remove
|
||||
# any the user previously uninstalled so the update doesn't
|
||||
# resurrect them.
|
||||
if api_v3.plugin_store_manager:
|
||||
try:
|
||||
purged = api_v3.plugin_store_manager.purge_uninstalled_plugins()
|
||||
if purged:
|
||||
logger.info(
|
||||
"Re-removed %d uninstalled plugin(s) restored by update: %s",
|
||||
len(purged), ", ".join(purged),
|
||||
)
|
||||
except (OSError, RuntimeError) as purge_err:
|
||||
logger.warning("Post-update plugin purge failed: %s", purge_err)
|
||||
else:
|
||||
logger.warning("git pull failed (returncode=%d): %s", result.returncode, result.stderr)
|
||||
# Show git's own first line: "check logs" leaves the user with
|
||||
# nothing to act on, and these failures are usually actionable
|
||||
# (conflicting local commits, no upstream, network).
|
||||
detail = next((ln.strip() for ln in (result.stderr or '').splitlines()
|
||||
if ln.strip()), '')
|
||||
pull_message = f"Update failed: {detail}" if detail else "Update failed; check logs for details"
|
||||
|
||||
# Nothing here restarts anything: the pull replaces files on
|
||||
# disk while the display and web services keep running the code
|
||||
# they loaded at boot. Without this the user is told the update
|
||||
# succeeded and sees no change until they happen to reboot.
|
||||
return jsonify({
|
||||
'status': 'success' if result.returncode == 0 else 'error',
|
||||
'message': pull_message,
|
||||
'restart_required': bool(result.returncode == 0 and code_changed),
|
||||
})
|
||||
elif action == 'checkout_branch':
|
||||
# Switch branches from the Tools tab. Needed because a checkout
|
||||
# that predates tracking (or a restored backup) can leave the pi
|
||||
# on a branch the update button cannot pull.
|
||||
result_payload, http_status = checkout_branch(
|
||||
str(PROJECT_ROOT), data.get('branch') or '', stash=bool(data.get('stash')))
|
||||
return jsonify(result_payload), http_status
|
||||
|
||||
elif action == 'restart_display_service':
|
||||
result = subprocess.run(['sudo', 'systemctl', 'restart', 'ledmatrix.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'restart_web_service':
|
||||
# Try to restart the web service (assuming it's ledmatrix-web.service)
|
||||
result = subprocess.run(['sudo', 'systemctl', 'restart', 'ledmatrix-web.service'],
|
||||
capture_output=True, text=True, timeout=10)
|
||||
elif action == 'install_base_requirements':
|
||||
# Base + web interface requirements: flask-compress and friends
|
||||
# live in web_interface/requirements.txt, not the root file.
|
||||
req_files = [f for f in (PROJECT_ROOT / 'requirements.txt',
|
||||
PROJECT_ROOT / 'web_interface' / 'requirements.txt')
|
||||
if f.exists()]
|
||||
if not req_files:
|
||||
return jsonify({'status': 'error', 'message': 'No requirements.txt found at project root'})
|
||||
outputs = []
|
||||
all_ok = True
|
||||
for req_file in req_files:
|
||||
label = req_file.relative_to(PROJECT_ROOT)
|
||||
# Isolate each file's install: a timeout or OSError on one
|
||||
# (e.g. requirements.txt) must not abort the rest of the
|
||||
# loop (e.g. web_interface/requirements.txt never attempted).
|
||||
try:
|
||||
result = _pip_install_requirements(req_file, timeout=120)
|
||||
all_ok = all_ok and result.returncode == 0
|
||||
outputs.append(f"== {label} ==\n" + _truncate_output(result.stdout, result.stderr))
|
||||
except subprocess.TimeoutExpired:
|
||||
all_ok = False
|
||||
outputs.append(f"== {label} ==\nTimed out after 120s")
|
||||
logger.warning("install_base_requirements timed out for %s", label)
|
||||
except OSError as install_err:
|
||||
all_ok = False
|
||||
outputs.append(f"== {label} ==\nFailed: {install_err}")
|
||||
logger.warning("install_base_requirements errored for %s: %s", label, install_err)
|
||||
return jsonify({
|
||||
'status': 'success' if all_ok else 'error',
|
||||
'message': 'Base requirements installed successfully' if all_ok else 'pip install failed',
|
||||
'output': "\n".join(outputs)
|
||||
})
|
||||
elif action == 'install_plugin_requirements':
|
||||
active_pm = getattr(api_v3, 'plugin_manager', None)
|
||||
if active_pm:
|
||||
plugins_dir = Path(active_pm.plugins_dir)
|
||||
else:
|
||||
_cm = getattr(api_v3, 'config_manager', None)
|
||||
_cfg = _cm.load_config() if _cm else {}
|
||||
_dir_name = _cfg.get('plugin_system', {}).get('plugins_directory', 'plugin-repos')
|
||||
plugins_dir = Path(_dir_name) if os.path.isabs(_dir_name) else PROJECT_ROOT / _dir_name
|
||||
results = []
|
||||
if plugins_dir.exists():
|
||||
for p in sorted(plugins_dir.iterdir()):
|
||||
req = p / 'requirements.txt'
|
||||
if p.is_dir() and req.exists():
|
||||
try:
|
||||
r = _pip_install_requirements(req, timeout=60)
|
||||
results.append({
|
||||
'plugin': p.name,
|
||||
'ok': r.returncode == 0,
|
||||
'output': _truncate_output(r.stdout, r.stderr)
|
||||
})
|
||||
except subprocess.TimeoutExpired:
|
||||
results.append({'plugin': p.name, 'ok': False, 'output': 'pip install timed out'})
|
||||
except OSError as exc:
|
||||
results.append({'plugin': p.name, 'ok': False, 'output': exc.strerror or 'OS error'})
|
||||
ok_count = sum(1 for r in results if r['ok'])
|
||||
all_ok = all(r['ok'] for r in results) if results else True
|
||||
return jsonify({
|
||||
'status': 'success' if all_ok else 'error',
|
||||
'message': f'Processed {len(results)} plugin(s) — {ok_count} succeeded' if results else 'No plugin requirements.txt files found',
|
||||
'details': results
|
||||
})
|
||||
elif action == 'force_git_reset':
|
||||
if not _GIT:
|
||||
return jsonify({'status': 'error', 'message': 'git not found on this system'}), 503
|
||||
project_dir = str(PROJECT_ROOT)
|
||||
fetch = subprocess.run(
|
||||
[_GIT, 'fetch', 'origin'],
|
||||
capture_output=True, text=True, timeout=30, cwd=project_dir
|
||||
)
|
||||
if fetch.returncode != 0:
|
||||
return jsonify({'status': 'error', 'message': 'git fetch failed', 'output': fetch.stderr.strip()})
|
||||
reset = subprocess.run(
|
||||
[_GIT, 'reset', '--hard', 'origin/main'],
|
||||
capture_output=True, text=True, timeout=30, cwd=project_dir
|
||||
)
|
||||
return jsonify({
|
||||
'status': 'success' if reset.returncode == 0 else 'error',
|
||||
'message': 'Reset to origin/main successfully' if reset.returncode == 0 else 'git reset failed',
|
||||
'output': (reset.stdout + reset.stderr).strip()
|
||||
})
|
||||
elif action == 'clear_pycache':
|
||||
cleared = 0
|
||||
failed = 0
|
||||
for d in PROJECT_ROOT.rglob('__pycache__'):
|
||||
if d.is_dir():
|
||||
try:
|
||||
shutil.rmtree(d)
|
||||
cleared += 1
|
||||
except OSError:
|
||||
failed += 1
|
||||
msg = f'Cleared {cleared} __pycache__ directories'
|
||||
if failed:
|
||||
msg += f' ({failed} could not be removed)'
|
||||
return jsonify({'status': 'success', 'message': msg})
|
||||
else:
|
||||
return jsonify({'status': 'error', 'message': 'Unknown action'}), 400
|
||||
|
||||
logger.info("system action '%s' returncode=%d", action, result.returncode)
|
||||
if result.returncode != 0 and result.stderr:
|
||||
logger.error("system action '%s' stderr: %s", action, result.stderr.strip())
|
||||
resp = {
|
||||
'status': 'success' if result.returncode == 0 else 'error',
|
||||
'message': 'Action completed' if result.returncode == 0 else 'Action failed; check logs for details',
|
||||
}
|
||||
if result.returncode != 0:
|
||||
resp['returncode'] = result.returncode
|
||||
resp['stderr'] = result.stderr.strip()
|
||||
return jsonify(resp)
|
||||
|
||||
except subprocess.TimeoutExpired as e:
|
||||
logger.error("system action '%s' timed out: %s", action, e)
|
||||
return jsonify({'status': 'error', 'message': 'Command timed out', 'returncode': -1, 'stderr': 'timeout'})
|
||||
except Exception as e:
|
||||
logger.error("execute_system_action failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'Action failed; see logs for details'}), 500
|
||||
@api_v3.route('/system/git-info', methods=['GET'])
|
||||
def get_git_info():
|
||||
"""Return branch, dirty state, recent commits and remote URL for the Tools tab."""
|
||||
if not _GIT:
|
||||
return jsonify({'status': 'error', 'message': 'git not found on this system'}), 503
|
||||
d = str(PROJECT_ROOT)
|
||||
try:
|
||||
branch = subprocess.run([_GIT, 'branch', '--show-current'], capture_output=True, text=True, timeout=10, cwd=d)
|
||||
if branch.returncode != 0:
|
||||
return jsonify({'status': 'error', 'message': f'git branch failed: {branch.stderr.strip()}'}), 500
|
||||
|
||||
status = subprocess.run([_GIT, 'status', '--short', '--untracked-files=no'], capture_output=True, text=True, timeout=15, cwd=d)
|
||||
if status.returncode != 0:
|
||||
return jsonify({'status': 'error', 'message': f'git status failed: {status.stderr.strip()}'}), 500
|
||||
|
||||
log = subprocess.run([_GIT, 'log', '--oneline', '-5'], capture_output=True, text=True, timeout=10, cwd=d)
|
||||
remote = subprocess.run([_GIT, 'remote', 'get-url', 'origin'], capture_output=True, text=True, timeout=10, cwd=d)
|
||||
branch_name = branch.stdout.strip()
|
||||
upstream = _git_upstream(d)
|
||||
return jsonify({
|
||||
'branch': branch_name,
|
||||
'dirty': bool(status.stdout.strip()),
|
||||
'status': status.stdout.strip(),
|
||||
'recent_commits': log.stdout.strip() if log.returncode == 0 else '',
|
||||
'remote_url': _scrub_git_remote_url(remote.stdout.strip()) if remote.returncode == 0 else '',
|
||||
# Surfaced so the Tools tab can warn before the user clicks Pull
|
||||
# Latest, rather than after it fails.
|
||||
'upstream': upstream,
|
||||
'can_pull': bool(upstream) or _git_remote_branch_exists(d, branch_name),
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error("get_git_info failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'Failed to get git info'}), 500
|
||||
@api_v3.route('/system/git-branches', methods=['GET'])
|
||||
def get_git_branches():
|
||||
"""List branches available to switch to, for the Tools tab picker."""
|
||||
if not _GIT:
|
||||
return jsonify({'status': 'error', 'message': 'git not found on this system'}), 503
|
||||
d = str(PROJECT_ROOT)
|
||||
try:
|
||||
# Refresh remote refs so a branch created since the last fetch shows up.
|
||||
subprocess.run([_GIT, 'fetch', 'origin', '--prune'],
|
||||
capture_output=True, text=True, timeout=60, cwd=d)
|
||||
|
||||
local = subprocess.run([_GIT, 'for-each-ref', '--format=%(refname:short)', 'refs/heads'],
|
||||
capture_output=True, text=True, timeout=15, cwd=d)
|
||||
remote = subprocess.run([_GIT, 'for-each-ref', '--format=%(refname:short)', 'refs/remotes/origin'],
|
||||
capture_output=True, text=True, timeout=15, cwd=d)
|
||||
if local.returncode != 0:
|
||||
return jsonify({'status': 'error', 'message': 'Could not list branches'}), 500
|
||||
|
||||
local_names = [b for b in local.stdout.split() if b]
|
||||
remote_names = []
|
||||
for ref in remote.stdout.split() if remote.returncode == 0 else []:
|
||||
name = ref.split('origin/', 1)[-1]
|
||||
# origin/HEAD is a symbolic alias, not a branch a user can pick.
|
||||
if name and name != 'HEAD' and name not in local_names:
|
||||
remote_names.append(name)
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'current': _git_current_branch(d),
|
||||
'upstream': _git_upstream(d),
|
||||
'local': sorted(local_names),
|
||||
'remote_only': sorted(remote_names),
|
||||
})
|
||||
except subprocess.TimeoutExpired:
|
||||
return jsonify({'status': 'error', 'message': 'Timed out talking to the remote'}), 504
|
||||
except OSError as e:
|
||||
logger.error("get_git_branches failed: %s", e, exc_info=True)
|
||||
return jsonify({'status': 'error', 'message': 'Failed to list branches'}), 500
|
||||
@@ -0,0 +1,361 @@
|
||||
"""Wi-Fi scanning, connection and status routes.
|
||||
|
||||
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
||||
endpoint names are unchanged by living here.
|
||||
"""
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
api_v3, describe_exception, jsonify, logger, request,
|
||||
)
|
||||
|
||||
|
||||
# WiFi Management Endpoints
|
||||
@api_v3.route('/wifi/status', methods=['GET'])
|
||||
def get_wifi_status():
|
||||
"""Get current WiFi connection status"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
status = wifi_manager.get_wifi_status()
|
||||
|
||||
# Get auto-enable setting from config
|
||||
auto_enable_ap = wifi_manager.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period)
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'connected': status.connected,
|
||||
'ssid': status.ssid,
|
||||
'ip_address': status.ip_address,
|
||||
'signal': status.signal,
|
||||
'ap_mode_active': status.ap_mode_active,
|
||||
'auto_enable_ap_mode': auto_enable_ap
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/scan', methods=['GET'])
|
||||
def scan_wifi_networks():
|
||||
"""Scan for available WiFi networks
|
||||
|
||||
If AP mode is active, it will be temporarily disabled during scanning
|
||||
and automatically re-enabled afterward. Users connected to the AP will
|
||||
be briefly disconnected during this process.
|
||||
"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
|
||||
# Check if AP mode is active before scanning (for user notification)
|
||||
ap_was_active = wifi_manager._is_ap_mode_active()
|
||||
|
||||
# Perform the scan (this will handle AP mode disabling/enabling internally)
|
||||
networks, _was_cached = wifi_manager.scan_networks()
|
||||
|
||||
# Convert to dict format
|
||||
networks_data = [
|
||||
{
|
||||
'ssid': net.ssid,
|
||||
'signal': net.signal,
|
||||
'security': net.security,
|
||||
'frequency': net.frequency
|
||||
}
|
||||
for net in networks
|
||||
]
|
||||
|
||||
response_data = {
|
||||
'status': 'success',
|
||||
'data': networks_data
|
||||
}
|
||||
|
||||
# Inform user if AP mode was temporarily disabled
|
||||
if ap_was_active:
|
||||
response_data['message'] = (
|
||||
f'Found {len(networks_data)} networks. '
|
||||
'Note: AP mode was temporarily disabled during scanning and has been re-enabled. '
|
||||
'If you were connected to the setup network, you may need to reconnect.'
|
||||
)
|
||||
|
||||
return jsonify(response_data)
|
||||
except Exception as e:
|
||||
logger.error("Error scanning WiFi networks", exc_info=True)
|
||||
error_message = 'An error occurred while scanning WiFi networks; see logs for details'
|
||||
|
||||
# Provide more specific error messages for common issues
|
||||
error_str = str(e).lower()
|
||||
if 'permission' in error_str or 'sudo' in error_str:
|
||||
error_message = (
|
||||
'Permission error while scanning. '
|
||||
'The WiFi scan requires appropriate permissions. '
|
||||
'Please ensure the application has necessary privileges.'
|
||||
)
|
||||
elif 'timeout' in error_str:
|
||||
error_message = (
|
||||
'WiFi scan timed out. '
|
||||
'The scan took too long to complete. '
|
||||
'This may happen if the WiFi interface is busy or in use.'
|
||||
)
|
||||
elif 'no wifi' in error_str or 'not available' in error_str:
|
||||
error_message = (
|
||||
'WiFi scanning tools are not available. '
|
||||
'Please ensure NetworkManager (nmcli) or iwlist is installed.'
|
||||
)
|
||||
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': error_message
|
||||
}), 500
|
||||
@api_v3.route('/wifi/connect', methods=['POST'])
|
||||
def connect_wifi():
|
||||
"""Connect to a WiFi network"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'Request body is required'
|
||||
}), 400
|
||||
|
||||
if 'ssid' not in data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'SSID is required'
|
||||
}), 400
|
||||
|
||||
ssid = data['ssid']
|
||||
if not ssid or not ssid.strip():
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'SSID cannot be empty'
|
||||
}), 400
|
||||
|
||||
ssid = ssid.strip()
|
||||
password = data.get('password', '') or ''
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
success, message = wifi_manager.connect_to_network(ssid, password)
|
||||
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message or 'Failed to connect to network'
|
||||
}), 400
|
||||
except Exception as e:
|
||||
logger.error("Error connecting to WiFi", exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/disconnect', methods=['POST'])
|
||||
def disconnect_wifi():
|
||||
"""Disconnect from the current WiFi network"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
success, message = wifi_manager.disconnect_from_network()
|
||||
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message or 'Failed to disconnect from network'
|
||||
}), 400
|
||||
except Exception as e:
|
||||
logger.error("Error disconnecting from WiFi", exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/ap/enable', methods=['POST'])
|
||||
def enable_ap_mode():
|
||||
"""Enable access point mode"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
_force_raw = (request.get_json(silent=True) or {}).get('force', False)
|
||||
force = _force_raw is True or (isinstance(_force_raw, str) and _force_raw.lower() in ('true', '1'))
|
||||
success, message = wifi_manager.enable_ap_mode(force=force)
|
||||
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message
|
||||
}), 400
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/ap/disable', methods=['POST'])
|
||||
def disable_ap_mode():
|
||||
"""Disable access point mode"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
success, message = wifi_manager.disable_ap_mode()
|
||||
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message
|
||||
}), 400
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/ap/auto-enable', methods=['GET'])
|
||||
def get_auto_enable_ap_mode():
|
||||
"""Get auto-enable AP mode setting"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
auto_enable = wifi_manager.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period)
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'auto_enable_ap_mode': auto_enable
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/ap/auto-enable', methods=['POST'])
|
||||
def set_auto_enable_ap_mode():
|
||||
"""Set auto-enable AP mode setting"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
data = request.get_json(silent=True)
|
||||
if data is None or 'auto_enable_ap_mode' not in data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'auto_enable_ap_mode is required'
|
||||
}), 400
|
||||
|
||||
auto_enable = bool(data['auto_enable_ap_mode'])
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
wifi_manager.config["auto_enable_ap_mode"] = auto_enable
|
||||
wifi_manager._save_config()
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': f'Auto-enable AP mode set to {auto_enable}',
|
||||
'data': {
|
||||
'auto_enable_ap_mode': auto_enable
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error("%s failed", request.path, exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/radio', methods=['GET'])
|
||||
def get_wifi_radio():
|
||||
"""Get current WiFi radio state (enabled/disabled) and wired-fallback status."""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
state = wifi_manager.get_wifi_radio_state()
|
||||
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': state
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error("Error getting WiFi radio state", exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
@api_v3.route('/wifi/radio', methods=['POST'])
|
||||
def set_wifi_radio():
|
||||
"""Turn the WiFi radio on or off.
|
||||
|
||||
Body: {"enabled": bool, "force": bool (optional)}. Disabling is refused
|
||||
unless Ethernet is connected or force=True, to avoid locking the user out
|
||||
of this web interface.
|
||||
"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
data = request.get_json(silent=True) or {}
|
||||
if 'enabled' not in data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'enabled is required'
|
||||
}), 400
|
||||
|
||||
# Parse defensively: bool("false") is True, so mirror the string-aware
|
||||
# coercion used for `force` — the endpoint is a public contract, not just
|
||||
# the shipped UI (which always sends real JSON booleans).
|
||||
_enabled_raw = data['enabled']
|
||||
enabled = _enabled_raw is True or (isinstance(_enabled_raw, str) and _enabled_raw.lower() in ('true', '1', 'yes'))
|
||||
_force_raw = data.get('force', False)
|
||||
force = _force_raw is True or (isinstance(_force_raw, str) and _force_raw.lower() in ('true', '1', 'yes'))
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
success, message, reason = wifi_manager.set_wifi_radio(enabled, force=force)
|
||||
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message,
|
||||
'data': wifi_manager.get_wifi_radio_state()
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': message,
|
||||
'reason': reason
|
||||
}), 400
|
||||
except Exception as e:
|
||||
logger.error("Error setting WiFi radio state", exc_info=True)
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
Reference in New Issue
Block a user