tusing/functional_threaded.py

## functional_threaded.py
#######################################################
####            NOW AT A GITHUB REPO               ####
####    https://github.com/tusing/unicorn_phat     ####
#######################################################

#!/usr/bin/env python

# Display a list of user-defined color bars;
# fill the remaining area with sparkles.
# Designed for the Unicorn pHAT.
# By tusing.


import unicornhat as unicorn
from random import randint
import time, math, colorsys
import os, sys, subprocess, threading

# Initialization
unicorn.set_layout(unicorn.AUTO)
unicorn.rotation(0)
unicorn.brightness(0.35) # Tune to your preferences.
width,height=unicorn.get_shape()

# Line number for where each function will begin
function_pos = {}
# Store values for multithreaded fetching functions.
function_values = {}

def main(display_function, bar_functions, time_limit=None):
    """ The main display function. Uses function_pos to assign parts of the display to
    bar functions and display functions.

    Args:
        display_function (func): A function intended to take up the majority of the HAT's
            display. Should limit display area with the use of function_pos.
        bar_functions (func): A list of single-row "bars". Again, assign position with the
            use of function_pos.
        time_limit (int): How long to wait before quitting (in seconds).
    """
    if bar_functions is not None:
        for index, bar_function in enumerate(bar_functions):
            function_pos[bar_function] = width - index - 1
        if display_function is not None:
            function_pos[display_function] = width - len(bar_functions) - 1
    else:
        function_pos[display_function] = width - 1

    threads = [threading.Thread(target=function) for function in function_pos.keys()]
    for thread in threads:
        thread.start()

    if time_limit is not None:
        time.sleep(time_limit)
        print("Time limit reached!")
        os._exit(3)


######################################################################
#######################                     ##########################
#######################    BAR FUNCTIONS    ##########################
#######################                     ##########################
######################################################################

#######################     INTERNET BAR    ##########################
def internet_color(update_rate=5):
    """ Color bar - tests internet connectivity. Displays white if connected;
    orange if not.

    Args:
        update_rate (float): seconds to wait before checking connectivity again
    """
    # Ping a Google DNS server to check for internet connectivity.
    while True:
        ping_response = subprocess.Popen(["/bin/ping", "-c1", "-w100", "8.8.8.8"], stdout=subprocess.PIPE).stdout.read()
        if "1 received" in str(ping_response):
            moving_pixel(function_pos[internet_color], (0, 255, 255), (255, 255, 255))
        else:
            moving_pixel(function_pos[internet_color], (255, 255, 255), (255, 127, 80))
        unicorn.show()
        time.sleep(update_rate)


def color_bar(position, color):
    """ Display a single, static bar of ```color``` in ```position```.

    Args:
        position (int): the width index at which to display the bar
        color (int tuple): (R, G, B) tuple of the RGB color to be displayed
    """
    for height_index in range(height):
        unicorn.set_pixel(position, height_index, *color)
    return

def moving_pixel(position, color, background, speed=0.1, direction="right"):
    """ Display a right-moving pixel of color ```color``` on a color bar with
    color ```background``` in position ```position.```

    Args:
        position (int): The width index at which to display the bar animation
        color (int tuple): (R, G, B) tuple of the moving pixel's color
        background (int tuple): (R, G, B) tuple of the background color
        speed (float): how often to wait between pixel movements
        direction (string, "left" or "right"): the direction the pixel
            should move, with "right" being towards the USB ports
    """
    for height_index in range(height):
        color_bar(position, background)
        if direction == "right":
            unicorn.set_pixel(position, height_index, *color)
        if direction == "left":
            unicorn.set_pixel(position, height - height_index - 1, *color)
        unicorn.show()
        time.sleep(speed)
    color_bar(position, background)
    unicorn.show()


######################################################################
#######################                     ##########################
#######################  FETCHER FUNCTIONS  ##########################
#######################                     ##########################
######################################################################

def load_fetcher(update_rate=5):
    """ Get the load of the system and modify the relevant dictionary
    with the new load value.

    Args:
        update_rate (float): seconds to wait before fetching load value
    """
    while True:
        function_values[load_fetcher] = os.getloadavg()[0]
        time.sleep(update_rate)

def random_color():
    """ Generate a random RGB color.
    Returns:
        int tuple: (R, G, B) values
    """
    r, g, b = randint(0, 255), randint(0, 255), randint(0, 255)
    return (r, g, b)


######################################################################
#######################                     ##########################
#######################  DISPLAY FUNCTIONS  ##########################
#######################                     ##########################
######################################################################

#######################    LOAD SPARKLES    ##########################
def load_sparkles(color_function=None, update_rate=5):
    """ Fill the rest of the area with randomly-positioned sparkles.
    Frequency of sparkling increases with load^2 (for load>1).

    Args:
        color_function (func): Define a custom function for the
            sparkles' color, instead of a random rainbow.
        update_rate (float): How often to refresh system load value (seconds).
    """

    color_function = random_color if color_function is None else color_function

    def random_pixel(color_function):
        """ Generate a randomly positioned pixel with the color returned
        by color_function.

        Args:
            color_function (func): Should return a (R,G,B) color value.
        """
        color = color_function()

        def random_position():
            """ Get the position of a random pixel bound by
            function_pos. """
            x = randint(0, function_pos[load_sparkles])
            y = randint(0, (height-1))
            return (x,y)
        selected_pixel = random_position()

        ''' Aesthetic: If the randomly generated pixel is currently lit,
        turn it off and try with a new pixel. Also works as sort of a

        population control on how many pixels will be lit. '''
        while sum(unicorn.get_pixel(*selected_pixel)) > 0:
            unicorn.set_pixel(*(selected_pixel + (0, 0, 0)))
            selected_pixel = random_position()
        unicorn.set_pixel(*(selected_pixel + color))
        return

    ''' Sparkle with a frequency based off of the computer's current
    load. Fetch load value every update_rate seconds.'''
    function_values[load_fetcher] = 1
    threading.Thread(target=load_fetcher).start()
    while True:
        tick = 1
        if function_values[load_fetcher] > 1:
            tick = 1/(function_values[load_fetcher]**2) if function_values[load_fetcher] < 12 else 1/144
        for i in range(int(update_rate/tick)):
            random_pixel(color_function)
            unicorn.show()
            time.sleep(tick)


#######################    LOAD RAINBOW    ##########################
def load_rainbow(update_rate=5):
    """ A lightly modified version of Pimeroni's "rainbow" example.
    Displays a moving rainbow of colors that increases with load.

    Args:
        update_rate (float): How often to update the load value (seconds).
    """

    i = 0.0
    offset = 30
    function_values[load_fetcher] = 1
    threading.Thread(target=load_fetcher).start()
    while True:
        load_function = function_values[load_fetcher]/10 if function_values[load_fetcher] <= 10 else 10
        for w in range(int(update_rate/0.01)):
            i = i + load_function
            for y in range(height):
                for x in range(function_pos[load_rainbow] + 1):
                    r = 0#x * 32
                    g = 0#y * 32
                    xy = x + y / 4
                    r = (math.cos((x+i)/2.0) + math.cos((y+i)/2.0)) * 64.0 + 128.0
                    g = (math.sin((x+i)/1.5) + math.sin((y+i)/2.0)) * 64.0 + 128.0
                    b = (math.sin((x+i)/2.0) + math.cos((y+i)/1.5)) * 64.0 + 128.0
                    r = max(0, min(255, r + offset))
                    g = max(0, min(255, g + offset))
                    b = max(0, min(255, b + offset))
                    unicorn.set_pixel(x,y,int(r),int(g),int(b))
            unicorn.show()
            time.sleep(0.01)


#######################     LOAD MATRIX     ##########################
def load_matrix(update_rate=5):
    """ A heavily modified version of Pimeroni's "cross" example.
    Speed increases with n*load^2.

    Args:
        update_rate (float): seconds to wait before updating load value
    """

    points = []
    edge_pixels = []
    class LightPoint:
        def __init__(self):
            self.direction = randint(1, 4)
            if self.direction == 1:
                self.x = randint(0, function_pos[load_matrix])
                self.y = 0
            elif self.direction == 2:
                self.x = 0
                self.y = randint(0, height - 1)
            elif self.direction == 3:
                self.x = randint(0, function_pos[load_matrix])
                self.y = height - 1
            else:
                self.x = function_pos[load_matrix] - 1
                self.y = randint(0, height - 1)
            self.colour = []
            for i in range(0, 3):
                self.colour.append(randint(100, 255))
            self.oldxy = (self.x, self.y)

    def update_positions():
        for point in points:
            # Any point already at an edge has been in display at this
            #   edge for ```tick``` seconds already, so we delete it.
            check_edges(point)
            point.oldxy = (point.x, point.y)
            if point.direction == 1:
                point.y += 1
            elif point.direction == 2:
                point.x += 1
            elif point.direction == 3:
                point.y -= 1
            else:
                point.x -= 1
            # Delete points that would cause a boundary violation after
            #   the above coordinate update.
            check_boundaries(point)
        unicorn.show()

    def plot_points():
        for point in points:
            unicorn.set_pixel(point.x, point.y, point.colour[0], point.colour[1], point.colour[2])
            unicorn.set_pixel(*(point.oldxy + (0, 0, 0)))
            unicorn.show()

    def check_edges(point):
        """ Deletes points that have reached an edge.
        Args:
            point (LightPoint): The point that has reached an edge.
        """
        if (point.x == function_pos[load_matrix] and point.direction is 2) \
            or (point.x == 0 and point.direction is 4) \
            or (point.y == 0 and point.direction is 3) \
            or (point.y == height - 1 and point.direction is 1):
                unicorn.set_pixel(point.x, point.y, 0, 0, 0)
                points.remove(point)

    def check_boundaries(point):
        """ Deletes points beyond allowed boundaries.

        Args:
            point (LightPoint): The point to check for boundary violations.
        """
        if (point.x > function_pos[load_matrix] and point.direction is 2) \
            or (point.x < 0 and point.direction is 4) \
            or (point.y < 0 and point.direction is 3) \
            or (point.y > height - 1 and point.direction is 1):
                if point in points:
                    points.remove(point)

    function_values[load_fetcher] = 1
    threading.Thread(target=load_fetcher).start()
    tick_func = lambda load: 0.5/(load**2) if load > 1 else 1/4
    max_points = function_pos[load_matrix]*height/3

    while True:
        tick = tick_func(function_values[load_fetcher]) if function_values[load_fetcher] < 12 else tick_func(12)
        for w in range(int(update_rate/tick)):
            if len(points) < max_points and randint(0, 5) > 1:
                points.append(LightPoint())
            update_positions()
            plot_points()
            time.sleep(tick)

if __name__ == "__main__":
    main(load_matrix, (internet_color,))
	#######################################################
	#### NOW AT A GITHUB REPO ####
	#### https://github.com/tusing/unicorn_phat ####
	#######################################################

	#!/usr/bin/env python

	# Display a list of user-defined color bars;
	# fill the remaining area with sparkles.
	# Designed for the Unicorn pHAT.
	# By tusing.


	import unicornhat as unicorn
	from random import randint
	import time, math, colorsys
	import os, sys, subprocess, threading

	# Initialization
	unicorn.set_layout(unicorn.AUTO)
	unicorn.rotation(0)
	unicorn.brightness(0.35) # Tune to your preferences.
	width,height=unicorn.get_shape()

	# Line number for where each function will begin
	function_pos = {}
	# Store values for multithreaded fetching functions.
	function_values = {}

	def main(display_function, bar_functions, time_limit=None):
	""" The main display function. Uses function_pos to assign parts of the display to
	bar functions and display functions.

	Args:
	display_function (func): A function intended to take up the majority of the HAT's
	display. Should limit display area with the use of function_pos.
	bar_functions (func): A list of single-row "bars". Again, assign position with the
	use of function_pos.
	time_limit (int): How long to wait before quitting (in seconds).
	"""
	if bar_functions is not None:
	for index, bar_function in enumerate(bar_functions):
	function_pos[bar_function] = width - index - 1
	if display_function is not None:
	function_pos[display_function] = width - len(bar_functions) - 1
	else:
	function_pos[display_function] = width - 1

	threads = [threading.Thread(target=function) for function in function_pos.keys()]
	for thread in threads:
	thread.start()

	if time_limit is not None:
	time.sleep(time_limit)
	print("Time limit reached!")
	os._exit(3)


	######################################################################
	####################### ##########################
	####################### BAR FUNCTIONS ##########################
	####################### ##########################
	######################################################################

	####################### INTERNET BAR ##########################
	def internet_color(update_rate=5):
	""" Color bar - tests internet connectivity. Displays white if connected;
	orange if not.

	Args:
	update_rate (float): seconds to wait before checking connectivity again
	"""
	# Ping a Google DNS server to check for internet connectivity.
	while True:
	ping_response = subprocess.Popen(["/bin/ping", "-c1", "-w100", "8.8.8.8"], stdout=subprocess.PIPE).stdout.read()
	if "1 received" in str(ping_response):
	moving_pixel(function_pos[internet_color], (0, 255, 255), (255, 255, 255))
	else:
	moving_pixel(function_pos[internet_color], (255, 255, 255), (255, 127, 80))
	unicorn.show()
	time.sleep(update_rate)


	def color_bar(position, color):
	""" Display a single, static bar of ```color``` in ```position```.

	Args:
	position (int): the width index at which to display the bar
	color (int tuple): (R, G, B) tuple of the RGB color to be displayed
	"""
	for height_index in range(height):
	unicorn.set_pixel(position, height_index, *color)
	return

	def moving_pixel(position, color, background, speed=0.1, direction="right"):
	""" Display a right-moving pixel of color ```color``` on a color bar with
	color ```background``` in position ```position.```

	Args:
	position (int): The width index at which to display the bar animation
	color (int tuple): (R, G, B) tuple of the moving pixel's color
	background (int tuple): (R, G, B) tuple of the background color
	speed (float): how often to wait between pixel movements
	direction (string, "left" or "right"): the direction the pixel
	should move, with "right" being towards the USB ports
	"""
	for height_index in range(height):
	color_bar(position, background)
	if direction == "right":
	unicorn.set_pixel(position, height_index, *color)
	if direction == "left":
	unicorn.set_pixel(position, height - height_index - 1, *color)
	unicorn.show()
	time.sleep(speed)
	color_bar(position, background)
	unicorn.show()




	######################################################################
	####################### ##########################
	####################### FETCHER FUNCTIONS ##########################
	####################### ##########################
	######################################################################

	def load_fetcher(update_rate=5):
	""" Get the load of the system and modify the relevant dictionary
	with the new load value.

	Args:
	update_rate (float): seconds to wait before fetching load value
	"""
	while True:
	function_values[load_fetcher] = os.getloadavg()[0]
	time.sleep(update_rate)

	def random_color():
	""" Generate a random RGB color.
	Returns:
	int tuple: (R, G, B) values
	"""
	r, g, b = randint(0, 255), randint(0, 255), randint(0, 255)
	return (r, g, b)



	######################################################################
	####################### ##########################
	####################### DISPLAY FUNCTIONS ##########################
	####################### ##########################
	######################################################################

	####################### LOAD SPARKLES ##########################
	def load_sparkles(color_function=None, update_rate=5):
	""" Fill the rest of the area with randomly-positioned sparkles.
	Frequency of sparkling increases with load^2 (for load>1).

	Args:
	color_function (func): Define a custom function for the
	sparkles' color, instead of a random rainbow.
	update_rate (float): How often to refresh system load value (seconds).
	"""

	color_function = random_color if color_function is None else color_function

	def random_pixel(color_function):
	""" Generate a randomly positioned pixel with the color returned
	by color_function.

	Args:
	color_function (func): Should return a (R,G,B) color value.
	"""
	color = color_function()

	def random_position():
	""" Get the position of a random pixel bound by
	function_pos. """
	x = randint(0, function_pos[load_sparkles])
	y = randint(0, (height-1))
	return (x,y)
	selected_pixel = random_position()

	''' Aesthetic: If the randomly generated pixel is currently lit,
	turn it off and try with a new pixel. Also works as sort of a

	population control on how many pixels will be lit. '''
	while sum(unicorn.get_pixel(*selected_pixel)) > 0:
	unicorn.set_pixel(*(selected_pixel + (0, 0, 0)))
	selected_pixel = random_position()
	unicorn.set_pixel(*(selected_pixel + color))
	return

	''' Sparkle with a frequency based off of the computer's current
	load. Fetch load value every update_rate seconds.'''
	function_values[load_fetcher] = 1
	threading.Thread(target=load_fetcher).start()
	while True:
	tick = 1
	if function_values[load_fetcher] > 1:
	tick = 1/(function_values[load_fetcher]**2) if function_values[load_fetcher] < 12 else 1/144
	for i in range(int(update_rate/tick)):
	random_pixel(color_function)
	unicorn.show()
	time.sleep(tick)


	####################### LOAD RAINBOW ##########################
	def load_rainbow(update_rate=5):
	""" A lightly modified version of Pimeroni's "rainbow" example.
	Displays a moving rainbow of colors that increases with load.

	Args:
	update_rate (float): How often to update the load value (seconds).
	"""

	i = 0.0
	offset = 30
	function_values[load_fetcher] = 1
	threading.Thread(target=load_fetcher).start()
	while True:
	load_function = function_values[load_fetcher]/10 if function_values[load_fetcher] <= 10 else 10
	for w in range(int(update_rate/0.01)):
	i = i + load_function
	for y in range(height):
	for x in range(function_pos[load_rainbow] + 1):
	r = 0#x * 32
	g = 0#y * 32
	xy = x + y / 4
	r = (math.cos((x+i)/2.0) + math.cos((y+i)/2.0)) * 64.0 + 128.0
	g = (math.sin((x+i)/1.5) + math.sin((y+i)/2.0)) * 64.0 + 128.0
	b = (math.sin((x+i)/2.0) + math.cos((y+i)/1.5)) * 64.0 + 128.0
	r = max(0, min(255, r + offset))
	g = max(0, min(255, g + offset))
	b = max(0, min(255, b + offset))
	unicorn.set_pixel(x,y,int(r),int(g),int(b))
	unicorn.show()
	time.sleep(0.01)


	####################### LOAD MATRIX ##########################
	def load_matrix(update_rate=5):
	""" A heavily modified version of Pimeroni's "cross" example.
	Speed increases with n*load^2.

	Args:
	update_rate (float): seconds to wait before updating load value
	"""

	points = []
	edge_pixels = []
	class LightPoint:
	def __init__(self):
	self.direction = randint(1, 4)
	if self.direction == 1:
	self.x = randint(0, function_pos[load_matrix])
	self.y = 0
	elif self.direction == 2:
	self.x = 0
	self.y = randint(0, height - 1)
	elif self.direction == 3:
	self.x = randint(0, function_pos[load_matrix])
	self.y = height - 1
	else:
	self.x = function_pos[load_matrix] - 1
	self.y = randint(0, height - 1)
	self.colour = []
	for i in range(0, 3):
	self.colour.append(randint(100, 255))
	self.oldxy = (self.x, self.y)

	def update_positions():
	for point in points:
	# Any point already at an edge has been in display at this
	# edge for ```tick``` seconds already, so we delete it.
	check_edges(point)
	point.oldxy = (point.x, point.y)
	if point.direction == 1:
	point.y += 1
	elif point.direction == 2:
	point.x += 1
	elif point.direction == 3:
	point.y -= 1
	else:
	point.x -= 1
	# Delete points that would cause a boundary violation after
	# the above coordinate update.
	check_boundaries(point)
	unicorn.show()

	def plot_points():
	for point in points:
	unicorn.set_pixel(point.x, point.y, point.colour[0], point.colour[1], point.colour[2])
	unicorn.set_pixel(*(point.oldxy + (0, 0, 0)))
	unicorn.show()

	def check_edges(point):
	""" Deletes points that have reached an edge.
	Args:
	point (LightPoint): The point that has reached an edge.
	"""
	if (point.x == function_pos[load_matrix] and point.direction is 2) \
	or (point.x == 0 and point.direction is 4) \
	or (point.y == 0 and point.direction is 3) \
	or (point.y == height - 1 and point.direction is 1):
	unicorn.set_pixel(point.x, point.y, 0, 0, 0)
	points.remove(point)

	def check_boundaries(point):
	""" Deletes points beyond allowed boundaries.

	Args:
	point (LightPoint): The point to check for boundary violations.
	"""
	if (point.x > function_pos[load_matrix] and point.direction is 2) \
	or (point.x < 0 and point.direction is 4) \
	or (point.y < 0 and point.direction is 3) \
	or (point.y > height - 1 and point.direction is 1):
	if point in points:
	points.remove(point)

	function_values[load_fetcher] = 1
	threading.Thread(target=load_fetcher).start()
	tick_func = lambda load: 0.5/(load**2) if load > 1 else 1/4
	max_points = function_pos[load_matrix]*height/3

	while True:
	tick = tick_func(function_values[load_fetcher]) if function_values[load_fetcher] < 12 else tick_func(12)
	for w in range(int(update_rate/tick)):
	if len(points) < max_points and randint(0, 5) > 1:
	points.append(LightPoint())
	update_positions()
	plot_points()
	time.sleep(tick)

	if __name__ == "__main__":
	main(load_matrix, (internet_color,))