mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 06:15:09 +00:00
refactor(web): one error-response path for api_v3 (#624)
* refactor(web): answer unhandled api_v3 errors from one blueprint handler
Fifty-three api_v3 routes ended in a copy of the same catch-all: log the
traceback, return {status, "An error occurred; see logs for details",
details: describe_exception(e)} with a 500. They are replaced by one
errorhandler on the api_v3 blueprint that returns exactly that body.
It lives on the blueprint rather than falling through to app.py's global
handler because the two answers differ: the global one adds
error_code: UNKNOWN_ERROR, and api_client.js sends a body with an
error_code to the error modal and one without to a plain toast. A
blueprint handler also gives tests that mount api_v3 on a bare Flask app
the same answer the real app gives.
Only handlers that were byte-for-byte that shape were removed (matched on
the AST, and each rewritten function re-parsed and compared). Handlers
with their own message, extra keys, operation-history records or cleanup
stay, as does execute_plugin_action's step-1 handler, which sits inside
an `except subprocess.TimeoutExpired` arm that would otherwise turn a
plugin's timeout into a 408.
HTTPExceptions raised inside a route go back as themselves in the global
handler's 4xx shape. Where a removed catch-all used to swallow one (only
delete_plugin_asset's non-silent get_json() is reachable), a malformed
request now gets its 415/400 instead of a 500.
Most of the diff is re-indentation from unwrapping the try blocks;
`git diff -w` shows the real change.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
* fix(web): plugin action errors name the real failure, not UnboundLocalError
execute_plugin_action bound a local `logger` in its JSON-parsing arm,
which made `logger` local to the whole function. Every other
`logger.error` in it then raised UnboundLocalError, so a failing OAuth
step-1 script was reported as "UnboundLocalError: cannot access local
variable 'logger'" -- from the step-1 handler, and before the previous
commit from the route's outer catch-all too. Use the module logger.
Found by comparing every api_v3 route's forced-failure response before
and after the catch-all consolidation.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
* refactor(web): drop the error category and exception-name code guessing
WebInterfaceError derived an ErrorCategory from every error code and put
it in each structured error body as `error_category`. Nothing reads it:
not the web UI (static/ and templates/), not the tests beyond the ones
pinning the mapping itself, and not any plugin in ledmatrix-plugins. The
enum, the inference table and the JSON key go.
from_exception() could also guess an error code from the exception's
class name ("Config" -> CONFIG_LOAD_FAILED, and so on). Every caller
passes a code, so the guess never ran; error_code is now required.
suggested_fixes stays: the error dialog in static/v3/js/utils/
error_handler.js lists them.
The REST reference loses error_category and says what an unanticipated
exception in an /api/v3 route answers.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
* refactor(web): one call for the from_exception error responses
Nine plugin routes built a structured error by hand:
from src.web_interface.errors import WebInterfaceError
error = WebInterfaceError.from_exception(e, ErrorCode.X)
return error_response(error.error_code, error.message,
details=error.details, context=error.context,
status_code=500)
That is now exception_error_response(e, ErrorCode.X) in api_helpers, so
error_response() is the only structured-error entry point the routes
use. The three operation-history routes never passed the context, and
with_context=False keeps their bodies exactly as they were; a test
compares the helper against the hand-written pair for both forms.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
* docs(changelog): one api_v3 error-response path
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
@@ -35,13 +35,16 @@ from typing import Dict, Any, Optional, Tuple, Type
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
logger = logging.getLogger(__name__)
|
||||
# Import new infrastructure
|
||||
from src.web_interface.api_helpers import success_response, error_response, validate_request_json
|
||||
from src.web_interface.api_helpers import (success_response, error_response,
|
||||
exception_error_response, validate_request_json)
|
||||
from src.web_interface.errors import ErrorCode
|
||||
from src.web_interface.secret_helpers import (find_secret_fields, mask_all_secret_values,
|
||||
merge_secrets, remove_empty_secrets,
|
||||
separate_secrets,
|
||||
strip_masked_values)
|
||||
from src.web_interface.error_handler import describe_exception, redact_text
|
||||
from src.web_interface.error_handler import (describe_exception, http_exception_payload,
|
||||
redact_text, unhandled_exception_payload)
|
||||
from werkzeug.exceptions import HTTPException
|
||||
from src.plugin_system.operation_types import OperationType
|
||||
from src.web_interface.validators import (
|
||||
validate_file_upload
|
||||
@@ -114,6 +117,43 @@ SYSTEM_FONTS = frozenset([
|
||||
'clr6x12', 'helvr12', 'texgyre-27'
|
||||
])
|
||||
api_v3 = Blueprint('api_v3', __name__)
|
||||
|
||||
|
||||
@api_v3.errorhandler(Exception)
|
||||
def _api_v3_unhandled_exception(error):
|
||||
"""The answer for any exception an api_v3 route does not handle itself.
|
||||
|
||||
Fifty-odd routes used to end in the same four lines -- log the traceback,
|
||||
return {status, message: "An error occurred; see logs for details",
|
||||
details: describe_exception(e)} with a 500. This is those four lines, once.
|
||||
A route still catches for itself when its failure needs something else: a
|
||||
specific message, extra keys, an operation-history record, or cleanup.
|
||||
|
||||
It is registered on the blueprint, not left to web_interface/app.py's
|
||||
global handler, because the two answers differ: the global one adds
|
||||
`error_code: UNKNOWN_ERROR`, and the plugin API client treats a body with
|
||||
an error_code differently from one without (see api_client.js). Tests that
|
||||
mount this blueprint on a bare Flask app get the same answer as the real
|
||||
app does, too.
|
||||
|
||||
`details` is describe_exception(), which redacts credentials and caps the
|
||||
length. CodeQL reads returning it as stack-trace exposure; it is the
|
||||
project's deliberate trade-off, because a device whose storage is failing
|
||||
otherwise answers "see logs for details" from the log viewer too
|
||||
(test_web_error_detail.py).
|
||||
|
||||
Werkzeug's HTTPExceptions subclass Exception, so a 400/405/413/415 raised
|
||||
inside a route lands here as well; it goes back as itself, in the global
|
||||
handler's shape. A 404 or explicit 500 never arrives: Flask prefers the
|
||||
app's code-specific handlers over a blueprint's class-based one.
|
||||
"""
|
||||
if isinstance(error, HTTPException):
|
||||
return jsonify(http_exception_payload(error)), error.code or 500
|
||||
logger.error("Unhandled exception in %s", request.endpoint or request.path,
|
||||
exc_info=error)
|
||||
return jsonify(unhandled_exception_payload(error)), 500
|
||||
|
||||
|
||||
def _get_plugin_version(plugin_id: str) -> str:
|
||||
"""Read the installed version from a plugin's manifest.json.
|
||||
|
||||
|
||||
@@ -47,15 +47,11 @@ def _day_setting(data, day, flat_key, nested_key):
|
||||
@api_v3.route('/config/main', methods=['GET'])
|
||||
def get_main_config():
|
||||
"""Get main configuration, with credentials redacted."""
|
||||
try:
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
|
||||
config = api_v3.config_manager.load_config()
|
||||
return jsonify({'status': 'success', 'data': _redact_credentials(config)})
|
||||
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
|
||||
config = api_v3.config_manager.load_config()
|
||||
return jsonify({'status': 'success', 'data': _redact_credentials(config)})
|
||||
@api_v3.route('/config/schedule', methods=['GET'])
|
||||
def get_schedule_config():
|
||||
"""Get current schedule configuration"""
|
||||
@@ -1165,20 +1161,16 @@ def save_main_config():
|
||||
@api_v3.route('/config/secrets', methods=['GET'])
|
||||
def get_secrets_config():
|
||||
"""Get secrets configuration"""
|
||||
try:
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
|
||||
config = api_v3.config_manager.get_raw_file_content('secrets')
|
||||
# This interface has no authentication, and this file is nothing but
|
||||
# credentials. It was handing all of them to anyone who could reach
|
||||
# the port. Values are masked; empty and YOUR_* placeholders are left
|
||||
# alone so a client can still tell "set" from "not set".
|
||||
return jsonify({'status': 'success',
|
||||
'data': mask_all_secret_values(config)})
|
||||
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
|
||||
config = api_v3.config_manager.get_raw_file_content('secrets')
|
||||
# This interface has no authentication, and this file is nothing but
|
||||
# credentials. It was handing all of them to anyone who could reach
|
||||
# the port. Values are masked; empty and YOUR_* placeholders are left
|
||||
# alone so a client can still tell "set" from "not set".
|
||||
return jsonify({'status': 'success',
|
||||
'data': mask_all_secret_values(config)})
|
||||
@api_v3.route('/config/raw/main', methods=['POST'])
|
||||
def save_raw_main_config():
|
||||
"""Save raw main configuration JSON"""
|
||||
|
||||
@@ -6,7 +6,7 @@ endpoint names are unchanged by living here.
|
||||
from web_interface.blueprints.api_v3 import (
|
||||
_ensure_display_service_running,
|
||||
_get_display_service_status, _stop_display_service, api_v3,
|
||||
describe_exception, jsonify, logger, os, request, uuid,
|
||||
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
|
||||
@@ -31,45 +31,41 @@ def _cache_manager():
|
||||
@api_v3.route('/display/current', methods=['GET'])
|
||||
def get_display_current():
|
||||
"""Get current display state"""
|
||||
import base64
|
||||
from PIL import Image
|
||||
import io
|
||||
|
||||
snapshot_path = "/tmp/led_matrix_preview.png"
|
||||
|
||||
# Get display dimensions from config: the logical size DisplayManager
|
||||
# renders at, so double-sided setups preview one screen
|
||||
from src.display_geometry import logical_size
|
||||
try:
|
||||
import base64
|
||||
from PIL import Image
|
||||
import io
|
||||
config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
|
||||
width, height = logical_size(config)
|
||||
except Exception:
|
||||
width, height = logical_size({})
|
||||
|
||||
snapshot_path = "/tmp/led_matrix_preview.png"
|
||||
|
||||
# Get display dimensions from config: the logical size DisplayManager
|
||||
# renders at, so double-sided setups preview one screen
|
||||
from src.display_geometry import logical_size
|
||||
# Try to read snapshot file
|
||||
image_data = None
|
||||
if os.path.exists(snapshot_path):
|
||||
try:
|
||||
config = api_v3.config_manager.load_config() if api_v3.config_manager else {}
|
||||
width, height = logical_size(config)
|
||||
except Exception:
|
||||
width, height = logical_size({})
|
||||
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
|
||||
|
||||
# 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
|
||||
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})
|
||||
@api_v3.route('/display/modes', methods=['GET'])
|
||||
def get_display_modes():
|
||||
"""Every display mode that can be requested on-demand, with its plugin.
|
||||
@@ -92,230 +88,208 @@ def get_display_modes():
|
||||
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
|
||||
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()
|
||||
# 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 {}
|
||||
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,
|
||||
})
|
||||
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
|
||||
return jsonify({'status': 'success', 'data': {'modes': modes}})
|
||||
@api_v3.route('/display/on-demand/status', methods=['GET'])
|
||||
def get_on_demand_status():
|
||||
"""Return the current on-demand display state."""
|
||||
try:
|
||||
cache = _cache_manager()
|
||||
# memory_ttl=0: the display service writes this key, so only the file
|
||||
# is current. This process's memory tier would keep serving the first
|
||||
# copy it read for the full max_age -- "active" for two minutes after
|
||||
# the display had already stopped.
|
||||
state = cache.get('display_on_demand_state', max_age=120, memory_ttl=0)
|
||||
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
|
||||
cache = _cache_manager()
|
||||
# memory_ttl=0: the display service writes this key, so only the file
|
||||
# is current. This process's memory tier would keep serving the first
|
||||
# copy it read for the full max_age -- "active" for two minutes after
|
||||
# the display had already stopped.
|
||||
state = cache.get('display_on_demand_state', max_age=120, memory_ttl=0)
|
||||
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
|
||||
}
|
||||
})
|
||||
@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)
|
||||
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
|
||||
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
|
||||
resolved_plugin = plugin_id
|
||||
resolved_mode = mode
|
||||
|
||||
if api_v3.plugin_manager:
|
||||
if resolved_plugin and resolved_plugin not in _pkg._discovered_plugin_manifests(resolved_plugin):
|
||||
return jsonify({'status': 'error', 'message': f'Plugin {resolved_plugin} not found'}), 404
|
||||
if api_v3.plugin_manager:
|
||||
if resolved_plugin and resolved_plugin not in _pkg._discovered_plugin_manifests(resolved_plugin):
|
||||
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:
|
||||
_pkg._discovered_plugin_manifests()
|
||||
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:
|
||||
_pkg._discovered_plugin_manifests()
|
||||
resolved_plugin = api_v3.plugin_manager.find_plugin_for_mode(resolved_mode)
|
||||
if not resolved_plugin:
|
||||
# Not among what was discovered: the plugin that declares
|
||||
# it may have been installed since. Scan once more.
|
||||
_pkg._discovered_plugin_manifests(rescan=True)
|
||||
resolved_plugin = api_v3.plugin_manager.find_plugin_for_mode(resolved_mode)
|
||||
if not resolved_plugin:
|
||||
# Not among what was discovered: the plugin that declares
|
||||
# it may have been installed since. Scan once more.
|
||||
_pkg._discovered_plugin_manifests(rescan=True)
|
||||
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
|
||||
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,
|
||||
)
|
||||
# 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 = _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)
|
||||
# Set the on-demand request in cache FIRST (before starting service)
|
||||
# This ensures the request is available when the service starts/restarts
|
||||
cache = _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)
|
||||
# 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 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")
|
||||
# Stop the display service first to ensure clean state when we will restart it
|
||||
if service_was_running and start_service:
|
||||
import 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:
|
||||
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': '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
|
||||
'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
|
||||
# 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
|
||||
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})
|
||||
@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)
|
||||
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 = _cache_manager()
|
||||
request_id = data.get('request_id') or str(uuid.uuid4())
|
||||
request_payload = {
|
||||
# Set the stop request in cache FIRST
|
||||
# The display controller will poll this and restart without the on-demand filter
|
||||
cache = _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,
|
||||
'action': 'stop',
|
||||
'timestamp': _pkg.time.time()
|
||||
'service': service_result
|
||||
}
|
||||
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.
|
||||
@@ -325,17 +299,13 @@ def get_current_display_status():
|
||||
System Logs page) can show what's on screen without querying the display
|
||||
process directly.
|
||||
"""
|
||||
try:
|
||||
cache = _cache_manager()
|
||||
# memory_ttl=0: written by the display service; see get_on_demand_status.
|
||||
state = cache.get('display_current_state', max_age=120, memory_ttl=0)
|
||||
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
|
||||
cache = _cache_manager()
|
||||
# memory_ttl=0: written by the display service; see get_on_demand_status.
|
||||
state = cache.get('display_current_state', max_age=120, memory_ttl=0)
|
||||
if state is None:
|
||||
state = {
|
||||
'mode': None,
|
||||
'plugin_id': None,
|
||||
'last_updated': None,
|
||||
}
|
||||
return jsonify({'status': 'success', 'data': state})
|
||||
|
||||
@@ -4,7 +4,7 @@ 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,
|
||||
PROJECT_ROOT, Path, Response, SYSTEM_FONTS, api_v3,
|
||||
jsonify, logger, os, re, request, validate_file_upload,
|
||||
)
|
||||
|
||||
@@ -52,439 +52,417 @@ def _catalog_response(catalog):
|
||||
@api_v3.route('/fonts/catalog', methods=['GET'])
|
||||
def get_fonts_catalog():
|
||||
"""Get fonts catalog"""
|
||||
# Check cache first (5 minute TTL)
|
||||
try:
|
||||
# Check cache first (5 minute TTL)
|
||||
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 _catalog_response(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
|
||||
|
||||
# BDF files are fixed-size bitmap strikes: FreeType
|
||||
# accepts only the pixel size baked into the file. The
|
||||
# UI needs to know that before offering a size control,
|
||||
# or it offers a number that cannot take effect.
|
||||
native_size = None
|
||||
if font_type == 'bdf':
|
||||
try:
|
||||
from src.element_style import _read_bdf_native_size
|
||||
native_size = _read_bdf_native_size(str(filepath))
|
||||
except Exception as e:
|
||||
logger.debug("Could not read native size for BDF font %s: %s",
|
||||
filepath, e)
|
||||
native_size = None
|
||||
|
||||
catalog[catalog_key] = {
|
||||
'filename': filename,
|
||||
'family_name': family_name,
|
||||
'display_name': display_name,
|
||||
'path': relative_path,
|
||||
'type': font_type,
|
||||
'is_system': is_system,
|
||||
'scalable': font_type != 'bdf',
|
||||
'native_size': native_size,
|
||||
'metadata': metadata if metadata else None
|
||||
}
|
||||
|
||||
# Cache the result (5 minute TTL) if available
|
||||
if set_cached:
|
||||
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 _catalog_response(cached_result)
|
||||
except ImportError:
|
||||
# Cache not available, continue without caching
|
||||
get_cached = None
|
||||
set_cached = None
|
||||
set_cached('fonts_catalog', catalog, ttl_seconds=300)
|
||||
except Exception:
|
||||
logger.error("[FontCatalog] Failed to cache fonts_catalog", exc_info=True)
|
||||
|
||||
# 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
|
||||
|
||||
# BDF files are fixed-size bitmap strikes: FreeType
|
||||
# accepts only the pixel size baked into the file. The
|
||||
# UI needs to know that before offering a size control,
|
||||
# or it offers a number that cannot take effect.
|
||||
native_size = None
|
||||
if font_type == 'bdf':
|
||||
try:
|
||||
from src.element_style import _read_bdf_native_size
|
||||
native_size = _read_bdf_native_size(str(filepath))
|
||||
except Exception as e:
|
||||
logger.debug("Could not read native size for BDF font %s: %s",
|
||||
filepath, e)
|
||||
native_size = None
|
||||
|
||||
catalog[catalog_key] = {
|
||||
'filename': filename,
|
||||
'family_name': family_name,
|
||||
'display_name': display_name,
|
||||
'path': relative_path,
|
||||
'type': font_type,
|
||||
'is_system': is_system,
|
||||
'scalable': font_type != 'bdf',
|
||||
'native_size': native_size,
|
||||
'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 _catalog_response(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
|
||||
return _catalog_response(catalog)
|
||||
@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
|
||||
# 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}})
|
||||
@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
|
||||
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
|
||||
font_file = request.files['font_file']
|
||||
if font_file.filename == '':
|
||||
return jsonify({'status': 'error', 'message': 'No file selected'}), 400
|
||||
|
||||
# Validate filename. validate_file_upload takes max_size_mb but only
|
||||
# checks the filename/extension with it -- it never looks at the
|
||||
# actual upload size, so the size limit below is enforced separately
|
||||
# before the file is saved (same pattern as the .star upload above).
|
||||
MAX_FONT_SIZE_MB = 10
|
||||
is_valid, error_msg = validate_file_upload(
|
||||
font_file.filename,
|
||||
max_size_mb=MAX_FONT_SIZE_MB,
|
||||
allowed_extensions=['.ttf', '.otf', '.bdf']
|
||||
)
|
||||
if not is_valid:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
# Check file size (stated limit is MAX_FONT_SIZE_MB)
|
||||
font_file.seek(0, 2) # Seek to end
|
||||
file_size = font_file.tell()
|
||||
font_file.seek(0) # Reset to beginning
|
||||
max_font_size_bytes = MAX_FONT_SIZE_MB * 1024 * 1024
|
||||
if file_size > max_font_size_bytes:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f'File too large (max {MAX_FONT_SIZE_MB}MB, got {file_size / 1024 / 1024:.1f}MB)'
|
||||
}), 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)
|
||||
# Validate filename. validate_file_upload takes max_size_mb but only
|
||||
# checks the filename/extension with it -- it never looks at the
|
||||
# actual upload size, so the size limit below is enforced separately
|
||||
# before the file is saved (same pattern as the .star upload above).
|
||||
MAX_FONT_SIZE_MB = 10
|
||||
is_valid, error_msg = validate_file_upload(
|
||||
font_file.filename,
|
||||
max_size_mb=MAX_FONT_SIZE_MB,
|
||||
allowed_extensions=['.ttf', '.otf', '.bdf']
|
||||
)
|
||||
if not is_valid:
|
||||
return jsonify({'status': 'error', 'message': error_msg}), 400
|
||||
|
||||
# Check file size (stated limit is MAX_FONT_SIZE_MB)
|
||||
font_file.seek(0, 2) # Seek to end
|
||||
file_size = font_file.tell()
|
||||
font_file.seek(0) # Reset to beginning
|
||||
max_font_size_bytes = MAX_FONT_SIZE_MB * 1024 * 1024
|
||||
if file_size > max_font_size_bytes:
|
||||
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
|
||||
'status': 'error',
|
||||
'message': f'File too large (max {MAX_FONT_SIZE_MB}MB, got {file_size / 1024 / 1024:.1f}MB)'
|
||||
}), 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}'
|
||||
})
|
||||
@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"""
|
||||
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:
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
import io
|
||||
import base64
|
||||
size = int(request.args.get('size', 12))
|
||||
except (ValueError, TypeError, OverflowError):
|
||||
return jsonify({'status': 'error', 'message': 'Invalid font size'}), 400
|
||||
|
||||
# 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)
|
||||
if not font_filename:
|
||||
return jsonify({'status': 'error', 'message': 'Font filename required'}), 400
|
||||
|
||||
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 size
|
||||
if size < 4 or size > 72:
|
||||
return jsonify({'status': 'error', 'message': 'Font size must be between 4 and 72'}), 400
|
||||
|
||||
# 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
|
||||
# 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
|
||||
|
||||
# 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
|
||||
# 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
|
||||
|
||||
if not font_filename:
|
||||
return jsonify({'status': 'error', 'message': 'Font filename required'}), 400
|
||||
# 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
|
||||
|
||||
# Validate size
|
||||
if size < 4 or size > 72:
|
||||
return jsonify({'status': 'error', 'message': 'Font size must be between 4 and 72'}), 400
|
||||
font_path = fonts_dir / safe_name
|
||||
|
||||
# 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
|
||||
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
|
||||
|
||||
# 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
|
||||
# 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
|
||||
|
||||
# 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
|
||||
if not font_path.exists():
|
||||
return jsonify({'status': 'error', 'message': f'Font file not found: {font_filename}'}), 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')
|
||||
# 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': '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
|
||||
'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
|
||||
}
|
||||
})
|
||||
@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
|
||||
# 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
|
||||
# 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
|
||||
# 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"
|
||||
# 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
|
||||
# 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
|
||||
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}"
|
||||
# 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
|
||||
# Security: Verify path is within fonts_dir
|
||||
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)
|
||||
potential_path.resolve().relative_to(fonts_dir.resolve())
|
||||
except ValueError:
|
||||
continue # Path escapes fonts_dir, skip
|
||||
|
||||
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
|
||||
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'
|
||||
})
|
||||
|
||||
@@ -186,13 +186,6 @@ def get_logs():
|
||||
'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():
|
||||
@@ -231,57 +224,49 @@ def get_sync_status():
|
||||
@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()
|
||||
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()
|
||||
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
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'cache_files': cache_files,
|
||||
'cache_dir': cache_dir,
|
||||
'total_files': len(cache_files)
|
||||
}
|
||||
})
|
||||
@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()
|
||||
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
|
||||
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']
|
||||
cache_key = data['key']
|
||||
|
||||
# The key names the file about to be removed. DiskCache refuses an
|
||||
# unusable key on its own, but silently: say so here instead of
|
||||
# reporting a deletion that never happened.
|
||||
if safe_path_component(cache_key) is None:
|
||||
return jsonify({'status': 'error', 'message': 'Invalid cache key'}), 400
|
||||
# The key names the file about to be removed. DiskCache refuses an
|
||||
# unusable key on its own, but silently: say so here instead of
|
||||
# reporting a deletion that never happened.
|
||||
if safe_path_component(cache_key) is None:
|
||||
return jsonify({'status': 'error', 'message': 'Invalid cache key'}), 400
|
||||
|
||||
# Delete the cache file
|
||||
api_v3.cache_manager.clear_cache(cache_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
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': f'Cache file for key "{cache_key}" deleted successfully'
|
||||
})
|
||||
def _errors_cache():
|
||||
"""The shared cache the display service publishes its errors to."""
|
||||
if not api_v3.cache_manager:
|
||||
|
||||
+1064
-1231
File diff suppressed because it is too large
Load Diff
@@ -24,95 +24,91 @@ import web_interface.blueprints.api_v3 as _pkg
|
||||
@api_v3.route('/system/status', methods=['GET'])
|
||||
def get_system_status():
|
||||
"""Get system status"""
|
||||
# Check cache first (10 second TTL for 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
|
||||
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
|
||||
# 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
|
||||
# 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
|
||||
# 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"
|
||||
# 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)
|
||||
# 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:
|
||||
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
|
||||
set_cached('system_status', status, ttl_seconds=10)
|
||||
except Exception:
|
||||
pass # Cache write failed, but continue
|
||||
|
||||
# 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
|
||||
return jsonify({'status': 'success', 'data': status})
|
||||
@api_v3.route('/system/version', methods=['GET'])
|
||||
def get_system_version():
|
||||
"""Get LEDMatrix repository version"""
|
||||
|
||||
@@ -111,34 +111,26 @@ def _parse_bool_ish(value):
|
||||
@api_v3.route('/wifi/status', methods=['GET'])
|
||||
def get_wifi_status():
|
||||
"""Get current WiFi connection status"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
status = wifi_manager.get_wifi_status()
|
||||
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)
|
||||
# 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,
|
||||
'last_connect_attempt': _last_connect_snapshot(),
|
||||
}
|
||||
})
|
||||
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
|
||||
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,
|
||||
'last_connect_attempt': _last_connect_snapshot(),
|
||||
}
|
||||
})
|
||||
@api_v3.route('/wifi/scan', methods=['GET'])
|
||||
def scan_wifi_networks():
|
||||
"""Scan for available WiFi networks
|
||||
@@ -219,241 +211,188 @@ def connect_wifi():
|
||||
background (see _last_connect_attempt); otherwise it waits for the result.
|
||||
"""
|
||||
global _last_connect_attempt
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
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()
|
||||
ap_mode_active = wifi_manager._is_ap_mode_active()
|
||||
|
||||
# One attempt at a time on either path: concurrent connects fight over
|
||||
# the radio, and the first to finish would clear the in-progress flag
|
||||
# the monitor daemon still needs for the other. The check can't depend
|
||||
# on AP state either -- a background attempt takes the AP down long
|
||||
# before it finishes.
|
||||
with _connect_lock:
|
||||
if _last_connect_attempt and _last_connect_attempt['state'] == 'pending':
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f"Already connecting to {_last_connect_attempt['ssid']}"
|
||||
}), 409
|
||||
_last_connect_attempt = {
|
||||
'ssid': ssid, 'state': 'pending', 'message': None,
|
||||
'error_type': None, 'finished_at': None,
|
||||
}
|
||||
|
||||
if ap_mode_active:
|
||||
try:
|
||||
_spawn(lambda: _run_background_connect(ssid, password))
|
||||
except Exception:
|
||||
# Nothing will ever finish this attempt; don't leave every
|
||||
# later request refused.
|
||||
with _connect_lock:
|
||||
_last_connect_attempt = None
|
||||
raise
|
||||
return jsonify({
|
||||
'status': 'pending',
|
||||
'message': (
|
||||
f'Connecting to {ssid}. The LEDMatrix-Setup network will turn off, '
|
||||
'so this page will lose its connection.'
|
||||
),
|
||||
'data': {'ssid': ssid},
|
||||
}), 202
|
||||
|
||||
try:
|
||||
success, message = wifi_manager.connect_to_network(ssid, password)
|
||||
except Exception as e:
|
||||
_record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)})
|
||||
raise
|
||||
payload = _connect_result_payload(ssid, success, message)
|
||||
_record_connect_result(ssid, payload)
|
||||
return jsonify(payload), (200 if success else 400)
|
||||
except Exception as e:
|
||||
logger.error("Error connecting to WiFi", exc_info=True)
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
'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()
|
||||
ap_mode_active = wifi_manager._is_ap_mode_active()
|
||||
|
||||
# One attempt at a time on either path: concurrent connects fight over
|
||||
# the radio, and the first to finish would clear the in-progress flag
|
||||
# the monitor daemon still needs for the other. The check can't depend
|
||||
# on AP state either -- a background attempt takes the AP down long
|
||||
# before it finishes.
|
||||
with _connect_lock:
|
||||
if _last_connect_attempt and _last_connect_attempt['state'] == 'pending':
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': f"Already connecting to {_last_connect_attempt['ssid']}"
|
||||
}), 409
|
||||
_last_connect_attempt = {
|
||||
'ssid': ssid, 'state': 'pending', 'message': None,
|
||||
'error_type': None, 'finished_at': None,
|
||||
}
|
||||
|
||||
if ap_mode_active:
|
||||
try:
|
||||
_spawn(lambda: _run_background_connect(ssid, password))
|
||||
except Exception:
|
||||
# Nothing will ever finish this attempt; don't leave every
|
||||
# later request refused.
|
||||
with _connect_lock:
|
||||
_last_connect_attempt = None
|
||||
raise
|
||||
return jsonify({
|
||||
'status': 'pending',
|
||||
'message': (
|
||||
f'Connecting to {ssid}. The LEDMatrix-Setup network will turn off, '
|
||||
'so this page will lose its connection.'
|
||||
),
|
||||
'data': {'ssid': ssid},
|
||||
}), 202
|
||||
|
||||
try:
|
||||
success, message = wifi_manager.connect_to_network(ssid, password)
|
||||
except Exception as e:
|
||||
_record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)})
|
||||
raise
|
||||
payload = _connect_result_payload(ssid, success, message)
|
||||
_record_connect_result(ssid, payload)
|
||||
return jsonify(payload), (200 if success else 400)
|
||||
@api_v3.route('/wifi/disconnect', methods=['POST'])
|
||||
def disconnect_wifi():
|
||||
"""Disconnect from the current WiFi network"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
success, message = wifi_manager.disconnect_from_network()
|
||||
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)
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
'message': message or 'Failed to disconnect from network'
|
||||
}), 400
|
||||
@api_v3.route('/wifi/ap/enable', methods=['POST'])
|
||||
def enable_ap_mode():
|
||||
"""Enable access point mode"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
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)
|
||||
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)
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
'message': message
|
||||
}), 400
|
||||
@api_v3.route('/wifi/ap/disable', methods=['POST'])
|
||||
def disable_ap_mode():
|
||||
"""Disable access point mode"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
success, message = wifi_manager.disable_ap_mode()
|
||||
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)
|
||||
if success:
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'message': message
|
||||
})
|
||||
else:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
'message': message
|
||||
}), 400
|
||||
@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
|
||||
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)
|
||||
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
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': {
|
||||
'auto_enable_ap_mode': auto_enable
|
||||
}
|
||||
})
|
||||
@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
|
||||
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 = _parse_bool_ish(data['auto_enable_ap_mode'])
|
||||
if auto_enable is None:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'auto_enable_ap_mode must be a boolean'
|
||||
}), 400
|
||||
|
||||
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)
|
||||
data = request.get_json(silent=True)
|
||||
if data is None or 'auto_enable_ap_mode' not in data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details',
|
||||
'details': describe_exception(e)
|
||||
}), 500
|
||||
'message': 'auto_enable_ap_mode is required'
|
||||
}), 400
|
||||
|
||||
auto_enable = _parse_bool_ish(data['auto_enable_ap_mode'])
|
||||
if auto_enable is None:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'auto_enable_ap_mode must be a boolean'
|
||||
}), 400
|
||||
|
||||
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
|
||||
}
|
||||
})
|
||||
@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
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
wifi_manager = WiFiManager()
|
||||
state = wifi_manager.get_wifi_radio_state()
|
||||
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
|
||||
return jsonify({
|
||||
'status': 'success',
|
||||
'data': state
|
||||
})
|
||||
@api_v3.route('/wifi/radio', methods=['POST'])
|
||||
def set_wifi_radio():
|
||||
"""Turn the WiFi radio on or off.
|
||||
@@ -462,53 +401,46 @@ def set_wifi_radio():
|
||||
unless Ethernet is connected or force=True, to avoid locking the user out
|
||||
of this web interface.
|
||||
"""
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
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 and a plain int never
|
||||
# matches `is True`, so `_parse_bool_ish` handles bool, string and
|
||||
# int 1/0 — the endpoint is a public contract, not just the shipped
|
||||
# UI (which always sends real JSON booleans). An unrecognized value
|
||||
# must be rejected, not silently disable the radio: this is the
|
||||
# route that can drop the caller's own connection to this interface.
|
||||
enabled = _parse_bool_ish(data['enabled'])
|
||||
if enabled is None:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'enabled must be a boolean'
|
||||
}), 400
|
||||
force = _parse_bool_ish(data.get('force', False))
|
||||
if force is None:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'force must be a boolean'
|
||||
}), 400
|
||||
|
||||
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)
|
||||
data = request.get_json(silent=True) or {}
|
||||
if 'enabled' not in data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
||||
}), 500
|
||||
'message': 'enabled is required'
|
||||
}), 400
|
||||
|
||||
# Parse defensively: bool("false") is True and a plain int never
|
||||
# matches `is True`, so `_parse_bool_ish` handles bool, string and
|
||||
# int 1/0 — the endpoint is a public contract, not just the shipped
|
||||
# UI (which always sends real JSON booleans). An unrecognized value
|
||||
# must be rejected, not silently disable the radio: this is the
|
||||
# route that can drop the caller's own connection to this interface.
|
||||
enabled = _parse_bool_ish(data['enabled'])
|
||||
if enabled is None:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'enabled must be a boolean'
|
||||
}), 400
|
||||
force = _parse_bool_ish(data.get('force', False))
|
||||
if force is None:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
'message': 'force must be a boolean'
|
||||
}), 400
|
||||
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user