fix asyncio calling, accept changes before mutating

effects.py

@@ -3,22 +3,22 @@ from colormath.color_conversions import convert_color
 import time
 import math
 
-def rainbow_whole(framecount, index, target):
+def hsv_to_target(hsv, target):
-    current_degree = framecount % 360
-    hsv = HSVColor(current_degree, 1, 1)
     rgb = [ x for x in convert_color(hsv, sRGBColor).get_upscaled_value_tuple() ]
     target[0] = rgb[0]
     target[1] = rgb[1]
     target[2] = rgb[2]
 
+def rainbow_whole(framecount, index, target):
+    current_degree = framecount % 360
+    hsv = HSVColor(current_degree, 1, 1)
+    hsv_to_target(hsv, target)
+
 def rainbow_whole_offset_stat(framecount, index, target):
     offset = index*90
     current_degree = (framecount + offset) % 360
     hsv = HSVColor(current_degree, 1, 1)
-    rgb = [ x for x in convert_color(hsv, sRGBColor).get_upscaled_value_tuple() ]
+    hsv_to_target(hsv, target)
-    target[0] = rgb[0]
-    target[1] = rgb[1]
-    target[2] = rgb[2]
 
 def rainbow_whole_offset_dyn(framecount, index, target):
     offset_hue_pixel = math.sin(framecount / 10000)*90*(index+1)
@@ -26,10 +26,7 @@ def rainbow_whole_offset_dyn(framecount, index, target):
     current_hue_degree = (framecount / 1000 + offset_hue_pixel) % 360
     current_sat        = (30+saturation_pixel) / 100
     hsv = HSVColor(current_hue_degree, current_sat, 1)
-    rgb = [ x for x in convert_color(hsv, sRGBColor).get_upscaled_value_tuple() ]
+    hsv_to_target(hsv, target)
-    target[0] = rgb[0]
-    target[1] = rgb[1]
-    target[2] = rgb[2]
 
 def two_colors(framecount, index, target, hue1, hue2):
     saturation_pixel = 1
@@ -38,10 +35,7 @@ def two_colors(framecount, index, target, hue1, hue2):
     current_hue_degree = hue_min + (hue_max - hue_min) / 2 + math.sin(framecount / 400 + index*100)*(hue_max - hue_min)/2
     current_sat = 1
     hsv = HSVColor(current_hue_degree, current_sat, 1)
-    rgb = [ x for x in convert_color(hsv, sRGBColor).get_upscaled_value_tuple() ]
+    hsv_to_target(hsv, target)
-    target[0] = rgb[0]
-    target[1] = rgb[1]
-    target[2] = rgb[2]
 
 def two_colors_red_blue(framecount, index, target):
     two_colors(framecount, index, target, 210, 359)
@@ -54,3 +48,9 @@ def two_colors_cyan_blue(framecount, index, target):
 
 def two_colors_cyan_orange(framecount, index, target):
     two_colors(framecount, index, target, 25, 180)
+
+def black(framecount, index, target):
+    target[0] = 0
+    target[1] = 0
+    target[2] = 0
+

main.py

@@ -57,25 +57,25 @@ async def unregister(websocket):
     await notify_users()
 
 
-async def counter(websocket, path):
+async def ws_handler(websocket, path):
     # register(websocket) sends user_event() to websocket
     await register(websocket)
     try:
-        await websocket.send(state_event(await state.get_data()))
+        await websocket.send(state_event(state.get_data()))
         await websocket.send(options_event())
         async for message in websocket:
             data = json.loads(message)
             if data["type"] == "mutation":
-                await state.mutate(data["payload"])
+                asyncio.create_task(state.propagate(data["payload"]))
             else:
                 logging.error("unsupported event: {}", data)
     finally:
         await unregister(websocket)
 
 async def serve_ws():
-    state.register_callback(notify_state)
+    state.register_on_change_callback(notify_state)
     await state.inc_initialized()
-    start_server = websockets.serve(counter, "localhost", 6789)
+    start_server = websockets.serve(ws_handler, "0.0.0.0", 6789)
     await start_server
 
 async def tick():

processor.py

@@ -6,7 +6,7 @@ from artnet import buildPacket
 from effects import *
 import state
 
-AVAILABLE_EFFECTS = [rainbow_whole, rainbow_whole_offset_stat, rainbow_whole_offset_dyn, two_colors_red_blue, two_colors_red_red, two_colors_cyan_blue, two_colors_cyan_orange ]
+AVAILABLE_EFFECTS = [ black, rainbow_whole, rainbow_whole_offset_stat, rainbow_whole_offset_dyn, two_colors_red_blue, two_colors_red_red, two_colors_cyan_blue, two_colors_cyan_orange]
 
 current_effect = rainbow_whole
 framecount = 0
@@ -21,6 +21,12 @@ def get_options():
         }
     }
 
+def serialize_internal_state():
+    global current_effect
+    return {
+        "current_effect": function_str(current_effect)
+    }
+
 def set_immidiate(framecount, channels, update):
     for inx, (current, target) in enumerate(zip(channels["current"], channels["target"])):
         update(framecount, inx, target)
@@ -35,6 +41,7 @@ async def handle_state_change(old, new):
         new_effect = [ x for x in AVAILABLE_EFFECTS if function_str(x) == new["current_effect"] ][0]
         current_effect = new_effect if new_effect != None else current_effect
         framecount = 0
+    await state.mutate(serialize_internal_state())
 
 async def main(target_ips):
     global current_effect
@@ -47,9 +54,9 @@ async def main(target_ips):
     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     debug = False
 
-    current_state = await state.get_data()
+    current_state = state.get_data()
-    state.register_callback(handle_state_change)
+    state.register_on_input_callback(handle_state_change)
-    await state.mutate({ "current_effect": function_str(current_effect) })
+    await state.mutate(serialize_internal_state())
     await state.inc_initialized()
 
     while True:

state.py

@@ -3,31 +3,41 @@ import copy
 from deepmerge import always_merger
 
 STATE = {"initialized": 0}
-state_lock = asyncio.Lock()
 
-NOTIFY_CALLBACKS = []
+ON_INPUT_CALLBACKS = []
+ON_CHANGE_CALLBACKS = []
 
-def register_callback(callback):
+def register_on_change_callback(callback):
-    global NOTIFY_CALLBACKS
+    global ON_CHANGE_CALLBACKS
-    NOTIFY_CALLBACKS = NOTIFY_CALLBACKS + [callback]
+    ON_CHANGE_CALLBACKS = ON_CHANGE_CALLBACKS + [callback]
 
-def unregister_callback(callback):
+def unregister_on_change_callback(callback):
-    NOTIFY_CALLBACK = [ x for x in NOTIFY_CALLBACKS if x != callback ]
+    ON_CHANGE_CALLBACK = [ x for x in ON_CHANGE_CALLBACKS if x != callback ]
+
+def register_on_input_callback(callback):
+    global ON_INPUT_CALLBACKS
+    ON_INPUT_CALLBACKS = ON_INPUT_CALLBACKS + [callback]
+
+def unregister_on_input_callback(callback):
+    ON_INPUT_CALLBACK = [ x for x in ON_INPUT_CALLBACKS if x != callback ]
 
 async def mutate(delta):
     global STATE
-    async with state_lock:
+    old_state = copy.deepcopy(STATE)
-        old_state = copy.deepcopy(STATE)
+    STATE = always_merger.merge(STATE, delta)
-        STATE = always_merger.merge(STATE, delta)
+    for callback in ON_CHANGE_CALLBACKS:
-        for callback in NOTIFY_CALLBACKS:
+        await callback(old_state, delta)
-            await callback(old_state, STATE)
 
-async def get_data():
+async def propagate(delta):
     global STATE
-    async with state_lock:
+    old_state = copy.deepcopy(STATE)
-        return STATE
+    for callback in ON_INPUT_CALLBACKS:
+        await callback(old_state, delta)
+
+def get_data():
+    global STATE
+    return STATE
 
 async def inc_initialized():
     global STATE
-    async with state_lock:
+    await mutate({"initialized": STATE["initialized"]+1 })
-        STATE["initialized"] += 1