diff --git a/main.py b/main.py
index 08970da1f05b7f56a04c888b93ac36b18a8e4c8c..09b520555790dbaa24ea0d7b7bc68e0791e82d88 100644
--- a/main.py
+++ b/main.py
@@ -16,6 +16,7 @@ from config import PROP_AVAILABLE_WIDGETS, PROP_AVAILABLE_PATTERNS, PROP_WINDOW_
PROP_FLASK_DEBUG, PROP_FLASK_HOST, PROP_FLASK_PORT, hw_config
from utils.jsonencoding import get_cfg_from_request, decode_dict, encode_dict
from utils.net import json_response, verify_json_request
+from utils.sequencing import validate_sequence
app = Flask(__name__)
@@ -133,7 +134,7 @@ def wr_set_widget():
# Return success
return json_response({
"status": "OK",
- "pattern": widget_name
+ "widget": widget_name
})
@app.route("/api/set_pattern", methods=["POST"])
@@ -169,6 +170,48 @@ def wr_set_pattern():
"pattern": pattern_name
})
+@app.route("/api/set_sequence", methods=["POST"])
+def wr_set_sequence():
+ # Make sure the request is valid
+ valid, validation_response = verify_json_request(request, ["widget", "pattern"])
+ if not valid:
+ return validation_response
+
+ # Make sure the widget and pattern sequences are valid
+ widget_seq = request.json.get("widget")
+ pattern_seq = request.json.get("pattern")
+ if (
+ (not isinstance(widget_seq, list)) or
+ (not isinstance(pattern_seq, list)) or
+ len(widget_seq) == 0 or
+ len(pattern_seq) == 0
+ ):
+ return json_response({
+ "error": "Invalid sequence provided"
+ }, 400)
+
+ # Make sure each widget is valid
+ for w_index in range(len(widget_seq)):
+ widget = widget_seq[w_index]
+ if not validate_sequence(widget, PROP_AVAILABLE_WIDGETS):
+ return json_response({
+ "error": "Invalid widget in sequence, index " + str(w_index)
+ }, 400)
+
+ # Make sure each pattern is valid
+ for p_index in range(len(pattern_seq)):
+ pattern = pattern_seq[p_index]
+ if not validate_sequence(pattern, PROP_AVAILABLE_PATTERNS):
+ return json_response({
+ "error": "Invalid pattern in sequence, index " + str(p_index)
+ }, 400)
+
+ # Set the sequences
+ mainthread_queue.put("SAVE_SEQUENCE_CFG:" + encode_dict(widget_seq) + ":" + encode_dict(pattern_seq))
+ return json_response({
+ "status": "OK"
+ })
+
@app.route("/api/get_config")
def wr_get_config():
return json_response(module_config)
@@ -347,6 +390,9 @@ if __name__ == "__main__":
current_widget_delta = 0
current_pattern_delta = 0
+ force_change_widget = False
+ force_change_pattern = False
+
# Fork off pygame
pygame_thread = threading.Thread(
target=lambda: run_screen(initial_widget),
@@ -382,51 +428,52 @@ if __name__ == "__main__":
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:
+ elif parts[0] == "SAVE_SEQUENCE_CFG" and len(parts) == 3:
widget_sequence = decode_dict(parts[1])
- with open("sequence_config.json", "w") as f:
- json.dump({
- "widget": widget_sequence,
- "pattern": 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({
- "widget": widget_sequence,
- "pattern": pattern_sequence
- }, f, indent=4)
+ pattern_sequence = decode_dict(parts[2])
+ if len(widget_sequence) <= current_widget_index:
+ current_widget_index = 0
+ if len(pattern_sequence) <= current_pattern_index:
+ current_pattern_index = 0
+ save_sequences()
+ elif message == "CYCLE_WIDGET":
+ force_change_widget = True
+ elif message == "CYCLE_PATTERN":
+ force_change_pattern = True
+
except queue.Empty:
pass
# Check if time delta has passed
# For Widgets
curr_widget = widget_sequence[current_widget_index]
- if curr_widget.get("duration", -1) != -1:
+ if curr_widget.get("duration", -1) != -1 or force_change_widget:
# Check if the widget has existed for too long
- if curr_widget.get("duration") * 1000 <= current_widget_delta:
+ if curr_widget.get("duration") * 1000 <= current_widget_delta or force_change_widget:
# Set the widget to the next widget
current_widget_index += 1
if current_widget_index >= len(widget_sequence):
current_widget_index = 0
display_queue.put("SET_WIDGET:" +
widget_sequence[current_widget_index].get("type") + ":" +
- encode_dict(widget_sequence[current_widget_index].get("config", {}))
+ encode_dict(widget_sequence[current_widget_index].get("config_override", {}))
)
current_widget_delta = 0
+ force_change_widget = False
# For patterns
curr_pattern = pattern_sequence[current_pattern_index]
- if curr_pattern.get("duration", -1) != -1:
+ if curr_pattern.get("duration", -1) != -1 or force_change_pattern:
# Check if the pattern has existed for too long
- if curr_pattern.get("duration") * 1000 <= current_pattern_delta:
+ if curr_pattern.get("duration") * 1000 <= current_pattern_delta or force_change_pattern:
# Set the pattern to the next pattern
current_pattern_index += 1
if current_pattern_index >= len(pattern_sequence):
current_pattern_index = 0
led_queue.put("SET_PATTERN:" +
pattern_sequence[current_pattern_index].get("type") + ":" +
- encode_dict(pattern_sequence[current_pattern_index].get("config", {}))
+ encode_dict(pattern_sequence[current_pattern_index].get("config_override", {}))
)
current_pattern_delta = 0
+ force_change_pattern = False
# Keep main thread alive
current_widget_delta += hw_config["main_update_frequency"]
time.sleep(hw_config["main_update_frequency"] / 1000)
diff --git a/utils/sequencing.py b/utils/sequencing.py
new file mode 100644
index 0000000000000000000000000000000000000000..4e904bc4210ae4d8415c6fd36999939a82fcbd05
--- /dev/null
+++ b/utils/sequencing.py
@@ -0,0 +1,23 @@
+def validate_sequence(seq: dict, available_seq_type: list[str]) -> bool:
+ """
+ Validates the entry sequence
+
+ All of these conditions must be met:
+ - seq is a dictionary
+ - seq has a "type" key
+ - seq["type"] is in available_seq_type
+ - seq has a "duration" key
+ - seq["duration"] is an integer or float
+ - seq["config_override"] is a dictionary or is not set
+ :param seq: the entry sequence
+ :param available_seq_type: the available sequence types
+ :return: True or False
+ """
+ return all([
+ isinstance(seq, dict),
+ "type" in seq,
+ seq.get("type") in available_seq_type,
+ "duration" in seq,
+ isinstance(seq.get("duration"), (int, float)),
+ isinstance(seq.get("config_override", {}), dict)
+ ])
\ No newline at end of file