Matrix/API/main.py

#!/usr/bin/env python

import os
from deps.samplebase import SampleBase
from rgbmatrix import graphics
from PIL import Image
import base64
import io

from flask import Flask, request, jsonify
from flask_cors import CORS, cross_origin
from dotenv import load_dotenv
from waitress import serve
from pathlib import Path
import time, threading


env_path = Path('..') / '.env'
load_dotenv(dotenv_path=env_path)

api = Flask(__name__)
cors = CORS(api)

@api.route('/upload', methods=['POST'])
@cross_origin()
def set_image():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking text and x,y coordinates 
		url = data.get('url')

		# call matrix method with data
		matrix.set_image(url)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/image', methods=['POST'])
@cross_origin()
def display_image():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking x,y coordinates 
		x_coord = int(data.get('x'))
		y_coord = int(data.get('y'))

		# call matrix method with data
		matrix.display_image(x_coord, y_coord)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/text', methods=['POST'])
@cross_origin()
def display_text():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking text and x,y coordinates 
		text = data.get('text')
		x_coord = int(data.get('x'))
		y_coord = int(data.get('y'))

		# call matrix method with data
		matrix.text(x_coord, y_coord, text)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/pixel', methods=['POST'])
@cross_origin()
def set_pixel():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking x,y coordinates 
		x_coord = int(data.get('x'))
		y_coord = int(data.get('y'))

		# call matrix method with data
		matrix.pixel(x_coord, y_coord)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/circle', methods=['POST'])
@cross_origin()
def draw_circle():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking radius and x,y coordinates 
		radius = int(data.get('r'))
		x_coord = int(data.get('x'))
		y_coord = int(data.get('y'))

		# call matrix method with data
		matrix.circle(x_coord, y_coord, radius)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/rectangle', methods=['POST'])
@cross_origin()
def draw_rectangle():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking rectangle width and height
		width = int(data.get('w'))
		height = int(data.get('h'))
		# try unpacking x,y coordinates 
		x_coord = int(data.get('x'))
		y_coord = int(data.get('y'))

		# call matrix method with data
		matrix.rectangle(x_coord, y_coord, width, height)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/line', methods=['POST'])
@cross_origin()
def draw_line():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking x,y coordinates 
		x1 = int(data.get('x1'))
		y1 = int(data.get('y1'))
		x2 = int(data.get('x2'))
		y2 = int(data.get('y2'))

		# call matrix method with data
		matrix.line(x1, y1, x2, y2)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/color', methods=['POST'])
@cross_origin()
def set_color():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# receive client data
		data = request.form
		# try unpacking red, green and blue color values 
		red = int(data.get('r'))
		green = int(data.get('g'))
		blue = int(data.get('b'))

		# call matrix method with data
		matrix.set_color(red, green, blue)
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/clear', methods=['GET'])
@cross_origin()
def clear_canvas():
	# prepare response data
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# clear matrix canvas
		matrix.clear()
	except Exception as e:
		# error handling
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)

@api.route('/update', methods=['GET'])
@cross_origin()
def update_matrix():
	response = { 'success': True }
	try:
		global matrix
		assert matrix is not None

		# swap matrix canvas
		matrix.update()
	except Exception as e:
		print(e)
		response['success'] = False

	# respond to client
	return jsonify(response)


# derived class to access rgb-led-matrix
class Matrix(SampleBase):
	def __init__(self, *args, **kwargs):
		super(Matrix, self).__init__(*args, **kwargs)
		self.parser.set_defaults(led_rows=int(os.getenv('PUBLIC_LED_WIDTH', 64)))
		self.parser.set_defaults(led_cols=int(os.getenv('PUBLIC_LED_HEIGHT', 64)))
		self.parser.set_defaults(led_chain=int(os.getenv('PUBLIC_LED_CHAIN', 1)))
		self.parser.set_defaults(led_parallel=int(os.getenv('PUBLIC_LED_PARALLEL', 1)))

	# run display program
	def run(self):
		# initialize font
		self.font = graphics.Font()
		self.font.LoadFont("deps/fonts/9x18B.bdf")
		# initialize color
		self.color = graphics.Color(255, 255, 255)
		# initialize image
		self.image = None
		
		# initialize canvas
		self.canvas = self.matrix.CreateFrameCanvas()
		self.canvas.Clear()

		# wait in idle
		while True:
			time.sleep(.05)

	# clear canvas
	def clear(self):
		self.canvas.Clear()

	# swap canvas with buffer
	def update(self):
		self.canvas = self.matrix.SwapOnVSync(self.canvas)
		self.clear()

	# set current color
	def set_color(self, r: int, g: int, b: int):
		self.color = graphics.Color(r, g, b)

	# set specified pixel to current color
	def pixel(self, x: int, y: int):
		self.canvas.SetPixel(x, y, self.color.red, self.color.green, self.color.blue)

	# display text at position with current color
	def text(self, x: int, y: int, t: int):
		graphics.DrawText(self.canvas, self.font, x*9, (y+1)*18, self.color, t)

	# display circle at position with radius in current color
	def circle(self, x: int, y: int, r: int):
		graphics.DrawCircle(self.canvas, x, y, r, self.color)

	# display rectangle at position with dimensions in current color
	def rectangle(self, x: int, y: int, w: int, h: int):
		# loop through each point in dimensions
		for i in range(w):
			for j in range(h):
				# set next pixel in rectangle
				self.canvas.SetPixel(x+i, y+j, self.color.red, self.color.green, self.color.blue)

	# display line from (x1,y1) to (x2,y2) in current color
	def line(self, x1: int, y1: int, x2: int, y2: int):
		graphics.DrawLine(self.canvas, x1, y1, x2, y2, self.color)

	# set current image
	def set_image(self, url: str):
		_, data = url.split(',')
		data = base64.b64decode(data)
		self.image = Image.open(io.BytesIO(data)).convert('RGB')

	# display image at position with dimensions in current color
	def display_image(self, x: int, y: int):
		if self.image is None: return
		# set image at position
		self.canvas.SetImage(self.image, x, y)


# initialize global matrix variable
matrix = None
if __name__ == "__main__":
	# set matrix to instance
	matrix = Matrix()
	# run rest api in thread, serving with 'waitress'
	threading.Thread(target=lambda: serve(api, host="0.0.0.0", port=8080)).start()

	# help menu on call
	if not matrix.process():
		matrix.print_help()