Flojoy snake demo

GAMES/GAMES_MATRIX/SNAKE_GAME/SNAKE_GAME.py

@@ -0,0 +1,185 @@
+import json
+import random
+from flojoy import flojoy, DataContainer
+import numpy as np
+
+from flojoy.small_memory import SmallMemory
+
+
+memory_key = "snake_game_info"
+snake_dimension = 22
+
+
+class SnakeData:
+    def __init__(
+        self,
+        node_id,
+        x_coords=[int(snake_dimension / 2)],
+        y_coords=[int(snake_dimension / 2)],
+        delta_x=0,
+        delta_y=1,
+        is_finished=False,
+        food_x=int(snake_dimension / 2),
+        food_y=snake_dimension - 4,
+        to_grow=False,
+    ) -> None:
+        self.node_id = node_id
+        self.x_coords = x_coords
+        self.y_coords = y_coords
+        self.delta_x = delta_x
+        self.delta_y = delta_y
+        self.is_finished = bool(is_finished)
+        self.food_x = food_x
+        self.food_y = food_y
+        self.to_grow = to_grow
+
+    def get_data(self):
+        return self.__dict__
+
+    @staticmethod
+    def from_data(node_id, data: dict):
+        if len(data) == 0:
+            return SnakeData(node_id)
+        snake_data = SnakeData(**data)
+        return snake_data
+
+    def print(self, prefix=""):
+        print(f"{prefix}snake Data:", json.dumps(self.get_data(), indent=2))
+
+
+@flojoy
+def SNAKE_GAME(dc_inputs: list[DataContainer], params: dict) -> dict:
+    """The SNAKE_GAME node is a specialized node that iterates through the body nodes for a given number of times.
+    To ensure proper functionality, the SNAKE_GAME node relies on a companion node called the `GOTO` node.
+    """
+
+    # get snake game data
+    control_input = dc_inputs[0]
+
+    if control_input.type != "ordered_pair":
+        raise ValueError(
+            f"unsupported DataContainer type passed for SNAKE_GAME: {control_input.type}"
+        )
+
+    node_id = params.get(
+        "node_id", 0
+    )  # WARNING: special case, it gets the node id from the params despite not being specified
+    print("NODE ID IS: ", node_id)
+    snake_data: SnakeData = load_data(node_id)
+
+    # get control inputs
+    delta_x = snake_data.delta_x
+    delta_y = snake_data.delta_y
+    if control_input.y[8]:
+        delta_x += -1
+        delta_y = 0
+    elif control_input.y[9]:
+        delta_x += 1
+        delta_y = 0
+    elif control_input.y[6]:
+        delta_y += -1
+        delta_x = 0
+    elif control_input.y[7]:
+        delta_y += 1
+        delta_x = 0
+
+    if delta_x + delta_y != 0:
+        snake_data.delta_x = int(delta_x / (abs(delta_x)) if delta_x != 0 else 0)
+        snake_data.delta_y = int(delta_y / (abs(delta_y)) if delta_y != 0 else 0)
+
+    # update snake position
+    snake_data.x_coords.insert(0, snake_data.x_coords[0] + snake_data.delta_x)
+    snake_data.y_coords.insert(0, snake_data.y_coords[0] + snake_data.delta_y)
+    if not snake_data.to_grow:
+        snake_data.x_coords.pop()
+        snake_data.y_coords.pop()
+    else:
+        snake_data.to_grow = False
+
+    # check if out of bounds
+    if (
+        snake_data.x_coords[0] < 0
+        or snake_data.x_coords[0] > snake_dimension - 1
+        or snake_data.y_coords[0] < 0
+        or snake_data.y_coords[0] > snake_dimension - 1
+    ):
+        snake_data.is_finished = True
+
+    # check if snake is eating itself
+    for i in range(len(snake_data.x_coords)):
+        if i == 0:
+            continue
+        if (
+            snake_data.x_coords[0] == snake_data.x_coords[i]
+            and snake_data.y_coords[0] == snake_data.y_coords[i]
+        ):
+            snake_data.is_finished = True
+
+    if snake_data.is_finished:
+        print("GAME OVER, RESETTING")
+        snake_data = SnakeData(node_id)
+        store_data(node_id, snake_data)
+        return output_game(snake_data)
+
+    # check if snake is eating food
+    if (
+        snake_data.x_coords[0] == snake_data.food_x
+        and snake_data.y_coords[0] == snake_data.food_y
+    ):
+        snake_data.to_grow = True
+        next_x, next_y = get_next_food_spot(snake_data)
+        snake_data.food_x = next_x
+        snake_data.food_y = next_y
+
+    # store snake game data
+    store_data(node_id, snake_data)
+
+    # output game
+    return output_game(snake_data)
+
+
+def load_data(node_id) -> SnakeData:
+    data = SmallMemory().read_memory(node_id, memory_key) or {}
+    print("LOAD DATA IS ", data)
+    snake_data = SnakeData.from_data(node_id=node_id, data=data)
+    return snake_data
+
+
+def store_data(node_id, snake_data: SnakeData):
+    SmallMemory().write_to_memory(node_id, memory_key, snake_data.get_data())
+    snake_data.print("store snake game data")
+
+
+def get_next_food_spot(snake_data):
+    taken = set()
+    for i in range(len(snake_data.x_coords)):
+        taken.add((snake_data.x_coords[i], snake_data.y_coords[i]))
+
+    free = set()
+    for i in range(snake_dimension):
+        for j in range(snake_dimension):
+            if (i, j) not in taken:
+                free.add((i, j))
+
+    return random.choice(list(free))
+
+
+# def delete_data(node_id):
+#     SmallMemory().delete_object(node_id, memory_key)
+#     print("delete snake game data")
+
+
+def output_game(snake_data):
+    r = [[10 for i in range(snake_dimension)] for j in range(snake_dimension)]
+    g = [[10 for i in range(snake_dimension)] for j in range(snake_dimension)]
+    b = [[10 for i in range(snake_dimension)] for j in range(snake_dimension)]
+    for i in range(len(snake_data.x_coords)):
+        x_coord = snake_data.x_coords[i]
+        y_coord = snake_data.y_coords[i]
+        g[y_coord][x_coord] = 255
+    r[snake_data.food_y][snake_data.food_x] = 255
+    dc = DataContainer(
+        type="image", r=np.array(r), g=np.array(g), b=np.array(b), a=None
+    )
+    print("OUTPUT GAME IS ", dc)
+    return dc

GAMES/GAMES_MATRIX/SNAKE_GAME/manifest.yml

@@ -0,0 +1,6 @@
+COMMAND:
+    - {
+        name: "Snake Game",
+        key: "SNAKE_GAME",
+        type: "GAME_MATRIX",
+      }

INSTRUMENTS/CONTROLLER/GAMEPAD/GAMEPAD.py

@@ -0,0 +1,79 @@
+from typing import Optional
+import hid
+import numpy as np
+from flojoy import OrderedPair, flojoy, DataContainer
+
+
+@flojoy
+def GAMEPAD(default: Optional[OrderedPair], params: dict) -> DataContainer:
+    """
+    The GAMEPAD node reads the input from a gamepad and returns a DataContainer with the following structure:
+    - it is of type ordered pair
+    - the x value indicates how many buttons are available
+    - the y value is a numpy array of booleans indicating which buttons are pressed
+    """
+
+    gamepad_device = None
+    clicked_buttons = [False] * 12
+
+    for device in hid.enumerate():
+        if "gamepad" in device["product_string"].lower():
+            gamepad_device = device
+
+    if gamepad_device is None:
+        print("ERROR: No gamepad found")
+        return OrderedPair(x=np.array([]), y=np.array(clicked_buttons))
+
+    gamepad = hid.device()
+    gamepad.open(gamepad_device["vendor_id"], gamepad_device["product_id"])
+
+    report = gamepad.read(64)
+
+    """
+    check backside buttons
+    """
+    if report[6] & 1:
+        clicked_buttons[0] = True
+    if report[6] & 2:
+        clicked_buttons[1] = True
+
+    """
+    check rightside buttons
+    """
+    if report[5] & 0b01000000:
+        clicked_buttons[2] = True
+
+    if report[5] & 0b00100000:
+        clicked_buttons[3] = True
+
+    if report[5] & 0b10000000:
+        clicked_buttons[4] = True
+
+    if report[5] & 0b00010000:
+        clicked_buttons[5] = True
+
+    """
+    check leftside buttons
+    """
+    if report[4] & 0b10000000:
+        clicked_buttons[6] = True
+
+    if report[4] == 0b00000000:
+        clicked_buttons[7] = True
+
+    if report[3] & 0b10000000:
+        clicked_buttons[8] = True
+
+    if report[3] == 0b00000000:
+        clicked_buttons[9] = True
+
+    """
+    check center buttons
+    """
+    if report[6] & 0b00010000:
+        clicked_buttons[10] = True
+
+if report[6] & 0b00100000:
+        clicked_buttons[11] = True
+
+    return OrderedPair(x=np.array([11]), y=np.array(clicked_buttons))

INSTRUMENTS/CONTROLLER/GAMEPAD/manifest.yml

@@ -0,0 +1,7 @@
+COMMAND:
+  - name: Gamepad 
+    key: GAMEPAD
+    type: CONTROLLER
+    pip_dependencies:
+      - name: hidapi
+        v: "0.14.0"

INSTRUMENTS/LED/LED_MATRIX_WS2812B/LED_MATRIX_WS2812B.py

@@ -0,0 +1,45 @@
+from flojoy import Image, flojoy, DataContainer
+import serial
+
+
+@flojoy
+def LED_MATRIX_WS2812B(default: Image, params: dict) -> dict:
+    """
+    The LED_MATRIX_WS2812B node takes an image as an input and sends signals to the LED Matrix to light up specific
+    LEDs accoring to the image input
+
+    Parameters:
+    - port: the port to which the LED Matrix is connected
+    - width: the width of the LED Matrix
+    - height: the height of the LED Matrix
+    """
+    input = default
+    port = params.get("port")
+    width = params.get("width")  # width of the LED matrix, not the image
+    height = params.get("height")
+
+    arduino = serial.Serial(port, 115200)
+
+    cmd = ""
+    # Use the function to light up the LED
+    for i in range(len(input.r)):
+        for j in range(len(input.r[0])):
+            # the LED matrix is a strand of LEDs that is folded into a matrix.
+            # So, for a 5x5 LED, we have 25 LEDs in consecutive order, and after the 5th LED
+            # on the first row, the 6th LED starts at the end of the second row, then after the 10th LED,
+            # the 11th LED starts at the beginning of the third row, and so on. Here is a drawing of how the
+            # LEDs are aligned:
+            if input.r[i][j] == 0 and input.g[i][j] == 0 and input.b[i][j] == 0:
+                continue
+
+            led_position = i * width
+            if i % 2 == 0:
+                led_position += j
+            else:
+                led_position += width - j - 1
+
+            cmd += f"{led_position},{input.r[i][j]},{input.g[i][j]},{input.b[i][j]},"
+
+    arduino.write((cmd[:-1] + "\n").encode())
+    # arduino.close()
+    return input  # return the input so that the next node can use it

INSTRUMENTS/LED/LED_MATRIX_WS2812B/manifest.yml

@@ -0,0 +1,17 @@
+COMMAND:
+  - name: Led Matrix
+    key: LED_MATRIX_WS2812B
+    type: LED
+    parameters:
+      port:
+        type: string
+        default: "/dev/ttyACM0"
+      width: 
+        type: int
+        default: 16
+      height: 
+        type: int
+        default: 16
+    pip_dependencies:
+      - name: pyserial
+        v: "3.5"