initial commit

main.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python
+# needed for websocket events
+import asyncio
+import websockets
+import json
+# needed for effect layer logic
+import png
+import numpy as np
+
+
+class Layer:
+    def __init__(self, new_path):
+        self.path = new_path
+        # read png file
+        self.png_reader = png.Reader(self.path)
+        # gray_scale_map_small = png_reader.read()
+        self.gray_scale_map_small_direct = self.png_reader.asRGB8()
+        # get png data from png object
+        self.png_width = self.gray_scale_map_small_direct[0]
+        self.png_height = self.gray_scale_map_small_direct[1]
+        self.png_image_data = self.gray_scale_map_small_direct[2]
+        # internal layer stuff
+        self.color_values = 3
+        self.rgb_buffer_length = (self.png_width * self.png_height) * self.color_values
+        self.layer_buffer_array = bytearray(self.rgb_buffer_length)
+        # just debug stuff
+        print("created layer for " + self.path)
+
+    def run(self):
+        print("next frame")
+
+    def print_png_values(self):
+        # get color values
+        for x in self.png_image_data:
+            print(x)
+
+    def make_brightness_image_filter(self):
+        img = self.png_image_data / 255
+        intensity = 255
+        intensity = intensity.astype(float)
+        for i in range(intensity.shape[2]):
+            intensity[:, :, i] *= img
+        intensity.astype(np.uint8)
+
+    def get_rgb_layer(self, r=0, g=0, b=0):
+        for i in range(0, self.rgb_buffer_length, self.color_values):
+            self.layer_buffer_array[i] = r
+            self.layer_buffer_array[i+1] = g
+            self.layer_buffer_array[i+2] = b
+
+
+# ------------------------------------
+# Websocket event part start
+# ------------------------------------
+def parse_json(input_json):
+    json_content = json.loads(input_json)
+    current_event = json_content["event"]
+    if current_event == "noteCut" or current_event == "noteFullyCut":
+        event_note_cut(json_content["noteCut"])
+    elif current_event == "bombCut":
+        event_bomb_cut()
+    elif current_event == "beatmapEvent":
+        event_beat_map(json_content["beatmapEvent"])
+    else:
+        print("other event: " + current_event)
+
+
+def event_note_cut(note_cut_object):
+    print("Note Cut")
+    event_note_cut_parse(note_cut_object)
+
+
+def event_note_cut_parse(note_cut_object):
+    saber_type = note_cut_object["saberType"]
+    note_type = note_cut_object["noteType"]
+    light_value = 0
+    if note_type == "NoteA":
+        light_value = 7
+    elif note_type == "NoteB":
+        light_value = 3
+
+    if saber_type == "SaberA":
+        trigger_saber_a(light_value)
+    elif saber_type == "SaberB":
+        trigger_saber_b(light_value)
+
+
+def trigger_saber_a(light_value):
+    print("saber a (red/left): " + str(light_value))
+
+
+def trigger_saber_b(light_value):
+    print("saber b (blue/right): " + str(light_value))
+
+
+def event_bomb_cut():
+    print("Bomb Cut")
+
+
+def event_beat_map(event_object):
+    # print("Beatmap")
+    event_beat_map_parse(event_object)
+
+
+def event_beat_map_parse(beatmap_event_object):
+    event_type = beatmap_event_object["type"]
+    event_value = beatmap_event_object["value"]
+    if 0 < event_type < 5:
+        if event_type == 0:
+            trigger_light_small(event_value)
+        elif event_type == 1:
+            trigger_light_big(event_value)
+        elif event_type == 2:
+            trigger_light_left(event_value)
+        elif event_type == 3:
+            trigger_light_right(event_value)
+        elif event_type == 4:
+            trigger_light_center(event_value)
+
+
+def trigger_light_small(value):
+    print("light small " + str(value))
+
+
+def trigger_light_big(value):
+    print("light big " + str(value))
+
+
+def trigger_light_center(value):
+    print("light center " + str(value))
+
+
+def trigger_light_left(value):
+    print("light left " + str(value))
+
+
+def trigger_light_right(value):
+    print("light right " + str(value))
+
+
+async def loop_websocket():
+    async with websockets.connect('ws://localhost:6557/socket') as websocket:
+        while True:
+            result = await websocket.recv()
+            # print(f"< {result}")
+            parse_json(result)
+
+# ------------------------------------
+# Websocket event part end
+# ------------------------------------
+
+if __name__ == '__main__':
+    layer_small = Layer('maps/small.png')
+    # layer_middle = Layer('maps/middle.png')
+    # layer_big = Layer('maps/big.png')
+    # layer_left = Layer('maps/left.png')
+    # layer_right = Layer('maps/right.png')
+
+    layer_small.print_png_values()
+
+    # Start websocket logic
+    asyncio.get_event_loop().run_until_complete(loop_websocket())