# This code needs to be located in the # rpi-rgb-led-matrix\bindings\python\samples directory to run. from samplebase import SampleBase import time, threading, sys class ManualColorFill(SampleBase): """ A class to remotely fill LED matrices/displays via GPIO inheriting from the SampleBase class. Attributes ---------- display_number : list The number of attached display matrices. color_settings : list The RGB color settings. LD_settings : list The light-dark cycle periods. display_status : list The current on/off status of a matrix. display_settings_count : list The number of settings of a matrix. action_time : list The next time the setting of a matrix is switched. action_signal : list The queue to contain the switch signals. x_position : list The x-positions of matrices in the array. y_position : list The y-positions of matrices in the array. x_dimension : list The x-dimensions of matrices in the canvas matrix. y_dimension : list The y-dimensions of matrices in the canvas matrix. Methods ------- display_params() Initializes the matrix settings. display_borders(display_number) Initializes the matrix pixel range. display_fill(display_number, color) Sets the pixels in a matrix to a RGB value. time_keeping(display_number) Recursively activates a timer thread for the next signal to a setting switch of a matrix. action_handling(display_number) Initiates matrix switching when a timer runs out. listen_for_key() Waits for an abort signal and shuts down if received. run() Starts the action loop, sets up a thread that listens for the abort signal, and shuts down if thread is killed. """ def __init__(self, *args, **kwargs): """ Parameters ---------- *args : str The parsed parameters from the remote machine. **kwargs : str The parsed parameters from the remote machine. Returns ------- None. """ super(ManualColorFill,self).__init__(*args, **kwargs) self.display_number = [] self.color_settings = [] self.LD_settings = [] self.display_status = [] self.display_settings_count = [] self.action_time = [] self.action_signal = [] self.x_position = [] self.y_position = [] self.x_dimension = [] self.y_dimension = [] def display_params(self): """ Initialize the display/matrix settings. Returns ------- None. """ for displays, colors, LD in self.lighting_settings: for display in displays: if display in self.display_number: i = self.display_number.index(display) self.color_settings[i].append(colors) self.LD_settings[i].append(LD) else: for display in displays: self.display_number.append(display) self.color_settings.append([colors]) self.LD_settings.append([LD]) self.display_status.append(0) self.display_settings_count.append(0) for display in self.display_number: self.display_borders(display) self.action_time.append(time.time()) def display_borders(self, display_number): """ Initialize the matrix pixel range. Parameters ---------- display_number : int The current display/matrix. Returns ------- None. """ if display_number % self.chain_length == 0: x_divisor = display_number / self.chain_length self.x_position.append(int(display_number / x_divisor -1)) else: self.x_position.append(display_number % self.chain_length -1) self.y_position.append((display_number -1) // self.chain_length) self.x_dimension.append([ int(self.matrix.width * self.x_position[ self.display_number.index(display_number) ] / self.chain_length), int(self.matrix.width * (self.x_position[ self.display_number.index(display_number) ] + 1) / self.chain_length)]) self.y_dimension.append([ int(self.matrix.height * self.y_position[ self.display_number.index(display_number)] / self.parallel), int(self.matrix.height * (self.y_position[ self.display_number.index(display_number) ] + 1) / self.parallel)]) def display_fill(self, display_number, color): """ Set the pixels in a matrix to a RGB value. Parameters ---------- display_number : int The current display/matrix. color : tuple RGB colors. Returns ------- None. """ d_index = self.display_number.index(display_number) for pixel_x in range( self.x_dimension[d_index][0], self.x_dimension[d_index][1]): for pixel_y in range( self.y_dimension[d_index][0], self.y_dimension[d_index][1]): self.matrix.SetPixel( pixel_x, pixel_y, color[0], color[1], color[2]) def time_keeping(self, display_number): """ Recursively activate a timer thread for the next signal to a setting switch of a matrix. Parameters ---------- display_number : int The current display/matrix. Returns ------- None. """ if self.action_time[ self.display_number.index(display_number)] > time.time(): timer_thread=threading.Timer( 0.1, self.time_keeping, (display_number,)) timer_thread.daemon=True timer_thread.start() else: self.action_signal.append(display_number) def action_handling(self, display_number): """ Initiate matrix switching when a timer ran out. Parameters ---------- display_number : int The current display/matrix. Returns ------- None. """ if self.display_status[self.display_number.index(display_number)] == 1: self.display_fill(display_number, [0,0,0]) self.display_status[self.display_number.index(display_number)] = 0 self.action_time[ self.display_number.index(display_number)] += self.LD_settings[ self.display_number.index(display_number) ][self.display_settings_count[ self.display_number.index(display_number)] ][1]*3600 if self.display_settings_count[ self.display_number.index(display_number) ] == len(self.color_settings[ self.display_number.index(display_number) ]) - 1: self.display_settings_count[ self.display_number.index(display_number) ] = 0 else: self.display_settings_count[ self.display_number.index(display_number) ] += 1 else: self.display_fill( display_number, self.color_settings[ self.display_number.index(display_number) ][self.display_settings_count[ self.display_number.index(display_number)]]) self.display_status[self.display_number.index(display_number)] = 1 self.action_time[ self.display_number.index(display_number) ] += self.LD_settings[ self.display_number.index(display_number) ][self.display_settings_count[ self.display_number.index(display_number)] ][0]*3600 self.time_keeping(display_number) def listen_for_key(self): """ Wait for an abort signal and shut down if received. Returns ------- None. """ while True: time.sleep(0.2) if input() == 'C': sys.exit() else: continue def run(self): """ Start the action loop, set up a thread that listens for the abort signal, and shut down if thread is killed. Raises ------ KeyboardInterrupt The KeyboardInterrupt is raised as it stems from a KeyboardInterrupt from the remote machine as a signal to shut down. Returns ------- None. """ death_thread = threading.Thread( target = self.listen_for_key, daemon=True) death_thread.start() time.sleep(0) self.display_params() for display in self.display_number: self.time_keeping(display) while True: if not death_thread.is_alive(): raise KeyboardInterrupt for display in self.action_signal: self.action_handling(display) self.action_signal = [] time.sleep(0.1) if __name__ == "__main__": manual_color_generator = ManualColorFill() if (not manual_color_generator.run()): manual_color_generator.print_help()