diff --git a/main.py b/main.py
index d95f3c8e4522dcf70db9c5a8a1f55f3c65fc1c20..818c2ff432058872e21d3fd781107b137f2f0379 100644
--- a/main.py
+++ b/main.py
@@ -6,7 +6,7 @@ import queue
import base64
import pygame
-from flask import request, Flask, Response, Request
+from flask import request, Flask
import ledpatterns
from absled import ILedString
@@ -14,6 +14,8 @@ import screenwidgets
from config import PROP_AVAILABLE_WIDGETS, PROP_AVAILABLE_PATTERNS, PROP_WINDOW_SCALE, PROP_WINDOW_FULLSCREEN, \
PROP_SCREEN_WIDTH, PROP_SCREEN_HEIGHT, PROP_HIDE_MOUSE, PROP_LEDS_DRIVER, PROP_LEDS_LENGTH, PROP_LEDS_LOCATION, \
PROP_FLASK_DEBUG, PROP_FLASK_HOST, PROP_FLASK_PORT, hw_config
+from utils.jsonencoding import get_cfg_from_request, decode_dict
+from utils.net import json_response, verify_json_request
app = Flask(__name__)
@@ -64,30 +66,17 @@ def load_sequences():
widget_sequence = seq.get("widgets", widget_sequence)
pattern_sequence = seq.get("patterns", pattern_sequence)
+def save_sequences():
+ with open("sequence_config.json", "w") as f:
+ json.dump({
+ "widgets": widget_sequence,
+ "patterns": pattern_sequence
+ }, f, indent=4)
+
mainthread_queue = queue.Queue()
display_queue = queue.Queue()
led_queue = queue.Queue()
-def json_response(data, status=200):
- return Response(json.dumps(data), status=status, mimetype="application/json")
-
-def verify_json_request(req: Request, required_fields=None):
- if required_fields is None:
- required_fields = []
-
- if not req.is_json:
- return False, json_response({
- "error": "Request is not JSON. Please make sure mimetype is set to application/json."
- }, 400)
-
- for field in required_fields:
- if field not in req.json:
- return False, json_response({
- "error": f"Missing field {field}"
- }, 400)
-
- return True, None
-
##############
# Web Routes #
@@ -133,13 +122,10 @@ def wr_set_widget():
"error": f"Widget {widget_name} is not available"
}, 400)
- config_override = request.json.get("config", {})
- if type(config_override) is not dict:
- return json_response({
- "error": "Config must be an Object"
- }, 400)
-
- config_b64 = base64.b64encode(json.dumps(config_override).encode("utf-8")).decode("utf-8")
+ # Get the config (or return error)
+ config_b64, config_error = get_cfg_from_request(request)
+ if config_error is not None:
+ return config_error
# Send the pattern to the LED thread
display_queue.put("SET_WIDGET:" + widget_name + ":" + config_b64)
@@ -169,13 +155,10 @@ def wr_set_pattern():
"error": f"Pattern {pattern_name} is not available"
}, 400)
- config_override = request.json.get("config", {})
- if type(config_override) is not dict:
- return json_response({
- "error": "Config must be an Object"
- }, 400)
-
- config_b64 = base64.b64encode(json.dumps(config_override).encode("utf-8")).decode("utf-8")
+ # Get the config (or return error)
+ config_b64, config_response = get_cfg_from_request(request)
+ if config_response is not None:
+ return config_response
# Send the pattern to the LED thread
led_queue.put("SET_PATTERN:" + pattern_name + ":" + config_b64)
@@ -377,10 +360,28 @@ if __name__ == "__main__":
while True:
try:
message = mainthread_queue.get_nowait()
+ parts = message.split(":")
if message == "EXIT":
display_queue.put("EXIT")
led_queue.put("EXIT")
break
+ elif parts[0] == "UPDATE_ALL_MODULE_CFG" and len(parts) == 2:
+ module_config = decode_dict(parts[1])
+ save_module_config()
+ elif parts[0] == "SAVE_WIDGET_SEQUENCE_CFG" and len(parts) == 2:
+ widget_sequence = decode_dict(parts[1])
+ with open("sequence_config.json", "w") as f:
+ json.dump({
+ "widgets": widget_sequence,
+ "patterns": pattern_sequence
+ }, f, indent=4)
+ elif parts[0] == "SAVE_PATTERN_SEQUENCE_CFG" and len(parts) == 2:
+ pattern_sequence = decode_dict(parts[1])
+ with open("sequence_config.json", "w") as f:
+ json.dump({
+ "widgets": widget_sequence,
+ "patterns": pattern_sequence
+ }, f, indent=4)
except queue.Empty:
pass
# Keep main thread alive
diff --git a/utils/jsonencoding.py b/utils/jsonencoding.py
new file mode 100644
index 0000000000000000000000000000000000000000..b9baf2a244e930e6e3edada5ff30e1a3abe1faa7
--- /dev/null
+++ b/utils/jsonencoding.py
@@ -0,0 +1,21 @@
+import base64
+import json
+
+from main import json_response
+
+
+def encode_dict(d: dict|list) -> str:
+ return base64.b64encode(json.dumps(d).encode("utf-8")).decode("utf-8")
+
+def decode_dict(s: str) -> dict|list:
+ return json.loads(base64.b64decode(s.encode("utf-8")).decode("utf-8"))
+
+def get_cfg_from_request(request):
+ config_override = request.json.get("config", {})
+ if type(config_override) is not dict:
+ return None, json_response({
+ "error": "Config must be an Object"
+ }, 400)
+
+ config_b64 = encode_dict(config_override)
+ return config_b64, None
\ No newline at end of file
diff --git a/utils/net.py b/utils/net.py
index ffc3f0e986255800d9c029771bf8cda9708c9676..8bdebb5560d39c656ba617fb5e25e29b0de622b3 100644
--- a/utils/net.py
+++ b/utils/net.py
@@ -1,7 +1,15 @@
+import json
import socket
+from typing import Any
+from flask import Response, Request
-def get_ip():
+
+def get_ip() -> str:
+ """
+ Gets the IP address of the current machine
+ :return: the IP address as a string
+ """
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0)
try:
@@ -12,4 +20,38 @@ def get_ip():
ip_addr = '127.0.0.1'
finally:
s.close()
- return ip_addr
\ No newline at end of file
+ return ip_addr
+
+
+def json_response(data: Any, status=200) -> Response:
+ """
+ Generate a JSON response from any json-serializable data
+ :param data: The data to be sent
+ :param status: The status code of the response
+ :return: A JSON response
+ """
+ return Response(json.dumps(data), status=status, mimetype="application/json")
+
+
+def verify_json_request(req: Request, required_fields: list|None = None) -> tuple[bool, Response|None]:
+ """
+ Verify that a request is JSON and contains all required fields
+ :param req: The request to verify
+ :param required_fields: A list of required fields, or None
+ :return: A tuple containing a boolean indicating if the request is valid and a response if it is not
+ """
+ if required_fields is None:
+ required_fields = []
+
+ if not req.is_json:
+ return False, json_response({
+ "error": "Request is not JSON. Please make sure mimetype is set to application/json."
+ }, 400)
+
+ for field in required_fields:
+ if field not in req.json:
+ return False, json_response({
+ "error": f"Missing field {field}"
+ }, 400)
+
+ return True, None
\ No newline at end of file