Changeset - 6e77747319ab
Parent rev.
Child rev.
[Not reviewed]
default
0 0 11
Matthew Reed (matthewreed) - 6 years ago 2019-04-30 17:46:33
matthewreed64@gmail.com
Added base software to select and run demos/games, along with a couple demos
11 files changed with 888 insertions and 0 deletions:
__init__.py
1
color_utils.py
86
controller.py
43
demos/__init__.py
1
demos/paint.py
108
demos/rainbow.py
69
demos/snake.py
191
lights.conf
5
lights.py
109
matrix.py
97
opc.py
178
0 comments (0 inline, 0 general)
__init__.py
➞
Show inline comments
 
new file 100644
 
color_utils.py
➞
Show inline comments
 
new file 100644
 
#!/usr/bin/env python
 
 
"""Helper functions to make color manipulations easier."""
 
 
from __future__ import division
 
import math
 
 
def remap(x, oldmin, oldmax, newmin, newmax):
 
    """Remap the float x from the range oldmin-oldmax to the range newmin-newmax
 
 
    Does not clamp values that exceed min or max.
 
    For example, to make a sine wave that goes between 0 and 256:
 
        remap(math.sin(time.time()), -1, 1, 0, 256)
 
 
    """
 
    zero_to_one = (x-oldmin) / (oldmax-oldmin)
 
    return zero_to_one*(newmax-newmin) + newmin
 
 
def clamp(x, minn, maxx):
 
    """Restrict the float x to the range minn-maxx."""
 
    return max(minn, min(maxx, x))
 
 
def cos(x, offset=0, period=1, minn=0, maxx=1):
 
    """A cosine curve scaled to fit in a 0-1 range and 0-1 domain by default.
 
 
    offset: how much to slide the curve across the domain (should be 0-1)
 
    period: the length of one wave
 
    minn, maxx: the output range
 
 
    """
 
    value = math.cos((x/period - offset) * math.pi * 2) / 2 + 0.5
 
    return value*(maxx-minn) + minn
 
 
def contrast(color, center, mult):
 
    """Expand the color values by a factor of mult around the pivot value of center.
 
 
    color: an (r, g, b) tuple
 
    center: a float -- the fixed point
 
    mult: a float -- expand or contract the values around the center point
 
 
    """
 
    r, g, b = color
 
    r = (r - center) * mult + center
 
    g = (g - center) * mult + center
 
    b = (b - center) * mult + center
 
    return (r, g, b)
 
 
def clip_black_by_luminance(color, threshold):
 
    """If the color's luminance is less than threshold, replace it with black.
 
 
    color: an (r, g, b) tuple
 
    threshold: a float
 
 
    """
 
    r, g, b = color
 
    if r+g+b < threshold*3:
 
        return (0, 0, 0)
 
    return (r, g, b)
 
 
def clip_black_by_channels(color, threshold):
 
    """Replace any individual r, g, or b value less than threshold with 0.
 
 
    color: an (r, g, b) tuple
 
    threshold: a float
 
 
    """
 
    r, g, b = color
 
    if r < threshold: r = 0
 
    if g < threshold: g = 0
 
    if b < threshold: b = 0
 
    return (r, g, b)
 
 
def mod_dist(a, b, n):
 
    """Return the distance between floats a and b, modulo n.
 
 
    The result is always non-negative.
 
    For example, thinking of a clock:
 
    mod_dist(11, 1, 12) == 2 because you can "wrap around".
 
 
    """
 
    return min((a-b) % n, (b-a) % n)
 
 
def gamma(color, gamma):
 
    """Apply a gamma curve to the color.  The color values should be in the range 0-1."""
 
    r, g, b = color
 
    return (max(r, 0) ** gamma, max(g, 0) ** gamma, max(b, 0) ** gamma)
controller.py
➞
Show inline comments
 
new file 100644
 
### controller.py
 
### Author: Matthew Reed
 
 
import sys
 
import time
 
import signal
 
import logging
 
import threading
 
import configparser
 
from enum import Enum
 
 
from evdev import InputDevice, categorize, ecodes
 
import asyncio
 
from select import select
 
 
class Controller:
 
 
    def __init__(self, config):
 
        self.config = config
 
        self.connect()
 
 
    def connect(self):
 
        try:
 
            self.dev = InputDevice('/dev/input/event0')
 
        except FileNotFoundError:
 
            self.dev = None
 
        except PermissionError:
 
            self.dev = None
 
 
    def read_input(self):
 
 
        if self.dev == None:
 
            self.connect()
 
 
        events = []
 
        if self.dev != None:
 
            try:
 
                for event in self.dev.read():
 
                    events.append(event)
 
            except BlockingIOError:
 
                pass
 
 
        return events
 
\ No newline at end of file
demos/__init__.py
➞
Show inline comments
 
new file 100644
 
__all__ = ['paint', 'rainbow', 'snake']
 
\ No newline at end of file
demos/paint.py
➞
Show inline comments
 
new file 100644
 
### paint.py
 
### Author: Matthew Reed
 
### Draw on the canvas using the D-Pad, A button changes color
 
 
import sys
 
import time
 
import signal
 
import logging
 
import configparser
 
from enum import Enum
 
 
import math
 
import matrix
 
 
class Paint:
 
 
    class DIRECTION(Enum):
 
        NONE = 0
 
        UP = 1
 
        DOWN = 2
 
        LEFT = 3
 
        RIGHT = 4
 
 
    def __init__(self, config, parent, matrix, controller):
 
        self.logger = logging.getLogger('paint')
 
        self.config = config
 
        self.parent = parent
 
        self.matrix = matrix
 
        self.controller = controller
 
 
    def reset(self):
 
        pass
 
 
    def splash(self):
 
        self.matrix.set_matrix((255,0,0))
 
        self.matrix.update()
 
 
    def run(self):
 
 
        pointer = 0
 
        direction = self.DIRECTION.NONE
 
        color = matrix.Colors.WHITE.value
 
        color_index = 7
 
 
        #start timers and counters
 
        self.start_time = time.time()
 
        last_time = time.time()
 
 
        led_iteration_count = 0
 
        frame_count = 0
 
 
        keep_going = True
 
 
        while keep_going:
 
 
            for event in self.controller.read_input():
 
                if event.code == 313 and event.value == 1:
 
                    keep_going = False
 
                elif event.code == 305 and event.value == 1:
 
                    color_index = (color_index + 1) % len(list(matrix.Colors))
 
                    color = list(matrix.Colors)[color_index].value
 
                elif event.code == 16:
 
                    if event.value == 1:
 
                        #dpad right
 
                        direction = self.DIRECTION.RIGHT
 
                    if event.value == 0:
 
                        #dpad none
 
                        direction = self.DIRECTION.NONE
 
                    if event.value == -1:
 
                        #dpad left
 
                        direction = self.DIRECTION.LEFT
 
                elif event.code == 17:
 
                    if event.value == 1:
 
                        #dpad down
 
                        direction = self.DIRECTION.DOWN
 
                    if event.value == 0:
 
                        #dpad none
 
                        direction = self.DIRECTION.NONE
 
                    if event.value == -1:
 
                        #dpad up
 
                        direction = self.DIRECTION.UP
 
 
            if time.time() > last_time + 0.1:
 
                last_time = time.time()
 
 
                pointerx = pointer % self.matrix.WIDTH
 
                pointery = math.floor(pointer / self.matrix.HEIGHT)
 
 
                if direction == self.DIRECTION.UP:
 
                    pointery = (pointery - 1) % self.matrix.HEIGHT
 
                elif direction == self.DIRECTION.DOWN:
 
                    pointery = (pointery + 1) % self.matrix.HEIGHT
 
                elif direction == self.DIRECTION.LEFT:
 
                    pointerx = (pointerx - 1) % self.matrix.WIDTH
 
                elif direction == self.DIRECTION.RIGHT:
 
                    pointerx = (pointerx + 1) % self.matrix.WIDTH
 
 
                pointer = pointery * self.matrix.HEIGHT + pointerx
 
 
                self.matrix.set_pixel(pointerx, pointery, color)
 
 
                self.matrix.update()
 
 
                led_iteration_count = (led_iteration_count + 1) % self.matrix.NUM_LEDS
 
                frame_count = frame_count + 1
 
 
            time.sleep(0.01)
 
 
\ No newline at end of file
demos/rainbow.py
➞
Show inline comments
 
new file 100644
 
### rainbow.py
 
### Author: Matthew Reed
 
### Rainbow array scrolls across the display. Rainbow generated using overlapping sin waves. The frequency, phase shift, center, and width all affect the output.
 
 
import sys
 
import time
 
import signal
 
import logging
 
import configparser
 
from enum import Enum
 
 
import math
 
 
class Rainbow:
 
 
    def __init__(self, config, parent, matrix, controller):
 
        self.logger = logging.getLogger('paint')
 
        self.config = config
 
        self.parent = parent
 
        self.matrix = matrix
 
        self.controller = controller
 
 
    def reset(self):
 
        pass
 
 
    def splash(self):
 
        self.matrix.set_matrix((0,255,0))
 
        self.matrix.update()
 
 
    def run(self):
 
 
        center = 128;
 
        width = 100;
 
        frequency = .1;
 
 
        #start timers and counters
 
        self.start_time = time.time()
 
        last_time = time.time()
 
 
        led_iteration_count = 0
 
        frame_count = 0
 
 
        keep_going = True
 
 
        while keep_going:
 
 
            for event in self.controller.read_input():
 
                if event.code == 313 and event.value == 1:
 
                    keep_going = False
 
 
            if time.time() > last_time + 0.1:
 
                last_time = time.time()
 
 
                for y in range(self.matrix.HEIGHT):
 
                    phaseShift = 2*math.pi - y*(1.5*math.pi/(self.matrix.HEIGHT-1))
 
                    for x in range(self.matrix.WIDTH):
 
                        i = (x + frame_count) % 128
 
                        red = math.sin(frequency*i + 0) * width + center;
 
                        green = math.sin(frequency*i + phaseShift/2) * width + center;
 
                        blue = math.sin(frequency*i + phaseShift) * width + center;
 
 
                        self.matrix.set_pixel(x, y, (red, green, blue))
 
 
                self.matrix.update()
 
 
                led_iteration_count = (led_iteration_count + 1) % self.matrix.NUM_LEDS
 
                frame_count = frame_count + 1
 
 
            time.sleep(0.01)
 
\ No newline at end of file
demos/snake.py
➞
Show inline comments
 
new file 100644
 
### snake.py
 
### Author: Matthew Reed
 
### Game of snake, uses the D-Pad
 
### Adapted from https://pythonspot.com/snake-with-pygame/ ###
 
 
import sys
 
import time
 
import signal
 
import logging
 
import configparser
 
from enum import Enum
 
 
import math
 
from random import randint
 
import matrix
 
 
class Snake:
 
 
    class Apple:
 
 
        def __init__(self, x, y):
 
            self.x = x
 
            self.y = y
 
 
        def draw(self, display):
 
            display.set_pixel(self.x, self.y, matrix.Colors.GREEN.value)
 
 
    class Player:
 
 
        class DIRECTION(Enum):
 
            NONE = 0
 
            UP = 1
 
            DOWN = 2
 
            LEFT = 3
 
            RIGHT = 4
 
 
        def __init__(self, length):
 
 
            self.length = length
 
            self.direction = self.DIRECTION.RIGHT
 
 
            self.updateCountMax = 4
 
            self.updateCount = 0
 
 
            # initial positions, no collision
 
            self.x = [2,1,0]
 
            self.y = [0,0,0]
 
 
            for i in range(0, 61):
 
                self.x.append(-100)
 
                self.y.append(-100)
 
 
        def update(self):
 
 
            self.updateCount = self.updateCount + 1
 
            if self.updateCount >= self.updateCountMax:
 
 
                # update previous positions
 
                for i in range(self.length, 0, -1):
 
                    self.x[i] = self.x[i-1]
 
                    self.y[i] = self.y[i-1]
 
 
                # update position of head of snake
 
                if self.direction == self.DIRECTION.RIGHT:
 
                    self.x[0] = self.x[0] + 1
 
                if self.direction == self.DIRECTION.LEFT:
 
                    self.x[0] = self.x[0] - 1
 
                if self.direction == self.DIRECTION.UP:
 
                    self.y[0] = self.y[0] - 1
 
                if self.direction == self.DIRECTION.DOWN:
 
                    self.y[0] = self.y[0] + 1
 
 
                self.updateCount = 0
 
 
        def is_overlaping(self, apple):
 
            for i in range(0, self.length):
 
                if apple.x == self.x[i] and apple.y == self.y[i]:
 
                    return True
 
            return False
 
 
        def draw(self, display):
 
            for i in range(0, self.length):
 
                #print("l: " + str(self.length) + " i: " + str(i) + " x: " + str(self.x[i]) + " y: " + str(self.y[i]))
 
                if self.x[i] > 7: self.x[i] = 7
 
                if self.x[i] < 0: self.x[i] = 0
 
                if self.y[i] > 7: self.y[i] = 7
 
                if self.y[i] < 0: self.y[i] = 0
 
                display.set_pixel(self.x[i], self.y[i], matrix.Colors.BLUE.value)
 
 
    def __init__(self, config, parent, matrix, controller):
 
        self.logger = logging.getLogger('snake')
 
        self.config = config
 
        self.parent = parent
 
        self.matrix = matrix
 
        self.controller = controller
 
 
    def reset(self):
 
        pass
 
 
    def splash(self):
 
        self.matrix.set_matrix(matrix.Colors.BLUE.value)
 
        self.matrix.update()
 
 
    def run(self):
 
 
        self.player = self.Player(3)
 
        self.apple = self.Apple(5,5)
 
 
        #start timers and counters
 
        self.start_time = time.time()
 
        last_time = time.time()
 
        delay_time = 0.2
 
 
        led_iteration_count = 0
 
        frame_count = 0
 
 
        keep_going = True
 
 
        while keep_going:
 
 
            for event in self.controller.read_input():
 
                if event.code == 313 and event.value == 1:
 
                    keep_going = False
 
                elif event.code == 16:
 
                    if event.value == 1:
 
                        #dpad right
 
                        self.player.direction = self.player.DIRECTION.RIGHT
 
                    if event.value == 0:
 
                        #dpad none
 
                        pass
 
                    if event.value == -1:
 
                        #dpad left
 
                        self.player.direction = self.player.DIRECTION.LEFT
 
                elif event.code == 17:
 
                    if event.value == 1:
 
                        #dpad down
 
                        self.player.direction = self.player.DIRECTION.DOWN
 
                    if event.value == 0:
 
                        #dpad none
 
                        pass
 
                    if event.value == -1:
 
                        #dpad up
 
                        self.player.direction = self.player.DIRECTION.UP
 
 
            if time.time() > last_time + delay_time:
 
                last_time = time.time()
 
 
                #determine next move
 
                self.player.update()
 
 
                #does snake eat apple?
 
                if self.apple.x == self.player.x[0] and self.apple.y == self.player.y[0]:
 
                    self.player.length = self.player.length + 1
 
                    while self.player.is_overlaping(self.apple):
 
                        self.apple.x = randint(0, 7)
 
                        self.apple.y = randint(0, 7)
 
                    #go faster as the snake gets longer
 
                    delay_time = delay_time - 0.005
 
 
                #does snake collide with itself?
 
                for i in range(1, self.player.length):
 
                    if self.player.x[0] == self.player.x[i] and self.player.y[0] == self.player.y[i]:
 
                        print("You lose! Collision: ")
 
                        print("x[0] (" + str(self.player.x[0]) + "," + str(self.player.y[0]) + ")")
 
                        print("x[" + str(i) + "] (" + str(self.player.x[i]) + "," + str(self.player.y[i]) + ")")
 
                        keep_going = False
 
 
                #does snake go off the board?
 
                if self.player.x[0] < 0 or self.player.x[0] > 7 or self.player.y[0] < 0 or self.player.y[0] > 7:
 
                    print("You lose! Off Board: ")
 
                    print("x[0] (" + str(self.player.x[0]) + "," + str(self.player.y[0]) + ")")
 
                    keep_going = False
 
 
                #update display
 
                self.matrix.set_matrix(matrix.Colors.OFF.value)
 
                self.player.draw(self.matrix)
 
                self.apple.draw(self.matrix)
 
                self.matrix.update()
 
 
                led_iteration_count = (led_iteration_count + 1) % self.matrix.NUM_LEDS
 
                frame_count = frame_count + 1
 
 
            time.sleep(0.01)
 
 
        #display score before exiting
 
        self.matrix.set_matrix(matrix.Colors.OFF.value)
 
        self.matrix.update()
 
        for i in range(0, self.player.length):
 
            self.matrix.set_pixel(i % self.matrix.WIDTH, math.floor(i / self.matrix.HEIGHT), matrix.Colors.WHITE.value)
 
        self.matrix.update()
 
        time.sleep(2)
 
\ No newline at end of file
lights.conf
➞
Show inline comments
 
new file 100644
 
[settings]
 
 
[leds]
 
num_channels = 1
 
num_leds = 64
 
\ No newline at end of file
lights.py
➞
Show inline comments
 
new file 100644
 
#!/usr/bin/python
 
 
### lights.py
 
### Author: Matthew Reed
 
 
import sys
 
import time
 
import signal
 
import logging
 
import threading
 
import configparser
 
from enum import Enum
 
 
import opc
 
import color_utils
 
import math
 
 
import matrix
 
import controller
 
from demos import *
 
 
 
class Lights:
 
 
    def __init__(self, config):
 
        self.logger = logging.getLogger('lights.lights')
 
        self.config = config
 
 
        self.matrix = matrix.Matrix(config)
 
        self.controller = controller.Controller(config)
 
 
        self.demo_i = 0
 
        self.demos = [
 
                paint.Paint(self.config, self, self.matrix, self.controller),
 
                rainbow.Rainbow(self.config, self, self.matrix, self.controller),
 
                snake.Snake(self.config, self, self.matrix, self.controller),
 
            ]
 
 
    def reset(self):
 
 
        self.color = matrix.Colors.WHITE.value
 
        self.matrix.set_matrix(matrix.Colors.OFF.value)
 
 
    def run(self):
 
 
        self.reset()
 
 
        #start timers and counters
 
        self.start_time = time.time()
 
        last_time = time.time()
 
 
        led_iteration_count = 0
 
        frame_count = 0
 
 
        while True:
 
 
            for event in self.controller.read_input():
 
                #print("Event: " + str(event))
 
                if event.code == 312 and event.value == 1:
 
                    self.demo_i = (self.demo_i + 1) % len(self.demos)
 
                    self.reset()
 
                elif event.code == 313 and event.value == 1:
 
                    self.reset()
 
                    self.demos[self.demo_i].run()
 
 
            if time.time() > last_time + 0.1:
 
                last_time = time.time()
 
 
                self.demos[self.demo_i].splash()
 
 
                led_iteration_count = (led_iteration_count + 1) % self.matrix.NUM_LEDS
 
                frame_count = frame_count + 1
 
 
            time.sleep(0.01)
 
 
 
    def stop(self):
 
        self.matrix.stop()
 
 
 
def main():
 
    logger = logging.getLogger('lights')
 
    logger.setLevel(logging.DEBUG)
 
    # create file handler which logs debug messages
 
    sh1 = logging.StreamHandler(sys.stdout)
 
    sh1.setLevel(logging.DEBUG)
 
    # create formatter and add it to the handlers
 
    formatter = logging.Formatter('[%(asctime)s][%(levelname)s][%(module)s][%(funcName)s] %(message)s')
 
    sh1.setFormatter(formatter)
 
    # add the handlers to the logger
 
    logger.addHandler(sh1)
 
    logger.info('Logger initialized')
 
 
    config = configparser.ConfigParser()
 
    config.read('lights.conf')
 
 
    try:
 
        my_lights = Lights(config)
 
 
        logger.info('Starting...')
 
        my_lights.run()
 
 
    finally:
 
        logger.info('Shutting down')
 
        my_lights.stop()
 
        logger.info('Goodbye')
 
 
if __name__ == "__main__":
 
    sys.exit(main())
 
\ No newline at end of file
matrix.py
➞
Show inline comments
 
new file 100644
 
### matrix.py
 
### Author: Matthew Reed
 
 
import sys
 
import time
 
import signal
 
import logging
 
import threading
 
import configparser
 
from enum import Enum
 
 
import opc
 
import color_utils
 
import math
 
 
 
class Colors(Enum):
 
    OFF = (0, 0, 0)
 
    RED = (255, 0, 0)
 
    GREEN = (0, 255, 0)
 
    BLUE = (0, 0, 255)
 
    PURPLE = (128, 0, 128)
 
    YELLOW = (255, 255, 0)
 
    ORANGE = (255, 140, 0)
 
    WHITE = (255, 255, 255)
 
    WHITE_LOW = (100, 100, 100)
 
 
class Matrix:
 
 
    def __init__(self, config):
 
        self.logger = logging.getLogger('matrix')
 
        self.config = config
 
 
        #init leds
 
        self.ADDRESS = 'localhost:7890'
 
        # Create a client object
 
        self.led_client = opc.Client(self.ADDRESS)
 
 
        # Test if it can connect (optional)
 
        if self.led_client.can_connect():
 
            self.logger.info('connected to FadeCandy %s' % self.ADDRESS)
 
        else:
 
            # We could exit here, but instead let's just print a warning
 
            # and then keep trying to send pixels in case the server
 
            # appears later
 
            self.logger.error('WARNING: could not connect to %s' % self.ADDRESS)
 
 
        self.num_channels = int(self.config.get('leds', 'num_channels'))
 
        self.NUM_LEDS = int(self.config.get('leds', 'num_leds'))
 
 
        self.my_pixels = [(0,0,0)] * self.NUM_LEDS
 
        self.led_client.put_pixels(self.my_pixels)
 
 
        #map out a matrix that represents the physical layout of the display
 
        self.led_map = [
 
            [63, 48, 47, 32, 31, 16, 15,  0],
 
            [62, 49, 46, 33, 30, 17, 14,  1],
 
            [61, 50, 45, 34, 29, 18, 13,  2],
 
            [60, 51, 44, 35, 28, 19, 12,  3],
 
            [59, 52, 43, 36, 27, 20, 11,  4],
 
            [58, 53, 42, 37, 26, 21, 10,  5],
 
            [57, 54, 41, 38, 25, 22,  9,  6],
 
            [56, 55, 40, 39, 24, 23,  8,  7],
 
        ]
 
 
        #x axis
 
        self.WIDTH = 8
 
        #y axis
 
        self.HEIGHT = 8
 
 
        #initialize matrix with zeros
 
        self.led_matrix = [[(0,0,0) for x in range(self.WIDTH)] for y in range(self.HEIGHT)]
 
 
    def update(self):
 
 
        for x in range(self.WIDTH):
 
            for y in range(self.HEIGHT):
 
                self.my_pixels[self.led_map[x][y]] = self.led_matrix[x][y]
 
 
        self.led_client.put_pixels(self.my_pixels)
 
 
    def set_pixel(self, x, y, color):
 
        self.led_matrix[x][y] = color
 
 
    def get_pixel(self, x, y):
 
        return self.led_matrix[x][y]
 
 
    def set_matrix(self, color):
 
        for x in range(self.WIDTH):
 
            for y in range(self.HEIGHT):
 
                self.led_matrix[x][y] = color
 
 
 
    def stop(self):
 
        self.logger.info('Turning off leds')
 
        self.my_pixels = [(0,0,0)] * self.NUM_LEDS
 
        self.led_client.put_pixels(self.my_pixels)
 
\ No newline at end of file
opc.py
➞
Show inline comments
 
new file 100644
 
#!/usr/bin/env python
 
 
"""Python Client library for Open Pixel Control
 
http://github.com/zestyping/openpixelcontrol
 
 
Sends pixel values to an Open Pixel Control server to be displayed.
 
http://openpixelcontrol.org/
 
 
Recommended use:
 
 
    import opc
 
 
    # Create a client object
 
    client = opc.Client('localhost:7890')
 
 
    # Test if it can connect (optional)
 
    if client.can_connect():
 
        print('connected to %s' % ADDRESS)
 
    else:
 
        # We could exit here, but instead let's just print a warning
 
        # and then keep trying to send pixels in case the server
 
        # appears later
 
        print('WARNING: could not connect to %s' % ADDRESS)
 
 
    # Send pixels forever at 30 frames per second
 
    while True:
 
        my_pixels = [(255, 0, 0), (0, 255, 0), (0, 0, 255)]
 
        if client.put_pixels(my_pixels, channel=0):
 
            print('...')
 
        else:
 
            print('not connected')
 
        time.sleep(1/30.0)
 
"""
 
 
import socket
 
import struct
 
import sys
 
 
class Client(object):
 
 
    def __init__(self, server_ip_port, long_connection=True, verbose=False):
 
        """Create an OPC client object which sends pixels to an OPC server.
 
 
        server_ip_port should be an ip:port or hostname:port as a single string.
 
        For example: '127.0.0.1:7890' or 'localhost:7890'
 
 
        There are two connection modes:
 
        * In long connection mode, we try to maintain a single long-lived
 
          connection to the server.  If that connection is lost we will try to
 
          create a new one whenever put_pixels is called.  This mode is best
 
          when there's high latency or very high framerates.
 
        * In short connection mode, we open a connection when it's needed and
 
          close it immediately after.  This means creating a connection for each
 
          call to put_pixels. Keeping the connection usually closed makes it
 
          possible for others to also connect to the server.
 
 
        A connection is not established during __init__.  To check if a
 
        connection will succeed, use can_connect().
 
 
        If verbose is True, the client will print debugging info to the console.
 
 
        """
 
        self.verbose = verbose
 
 
        self._long_connection = long_connection
 
 
        self._ip, self._port = server_ip_port.split(':')
 
        self._port = int(self._port)
 
 
        self._socket = None  # will be None when we're not connected
 
 
    def _debug(self, m):
 
        if self.verbose:
 
            print('    %s' % str(m))
 
 
    def _ensure_connected(self):
 
        """Set up a connection if one doesn't already exist.
 
 
        Return True on success or False on failure.
 
 
        """
 
        if self._socket:
 
            self._debug('_ensure_connected: already connected, doing nothing')
 
            return True
 
 
        try:
 
            self._debug('_ensure_connected: trying to connect...')
 
            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
            self._socket.connect((self._ip, self._port))
 
            self._debug('_ensure_connected:    ...success')
 
            return True
 
        except socket.error:
 
            self._debug('_ensure_connected:    ...failure')
 
            self._socket = None
 
            return False
 
 
    def disconnect(self):
 
        """Drop the connection to the server, if there is one."""
 
        self._debug('disconnecting')
 
        if self._socket:
 
            self._socket.close()
 
        self._socket = None
 
 
    def can_connect(self):
 
        """Try to connect to the server.
 
 
        Return True on success or False on failure.
 
 
        If in long connection mode, this connection will be kept and re-used for
 
        subsequent put_pixels calls.
 
 
        """
 
        success = self._ensure_connected()
 
        if not self._long_connection:
 
            self.disconnect()
 
        return success
 
 
    def put_pixels(self, pixels, channel=0):
 
        """Send the list of pixel colors to the OPC server on the given channel.
 
 
        channel: Which strand of lights to send the pixel colors to.
 
            Must be an int in the range 0-255 inclusive.
 
            0 is a special value which means "all channels".
 
 
        pixels: A list of 3-tuples representing rgb colors.
 
            Each value in the tuple should be in the range 0-255 inclusive.
 
            For example: [(255, 255, 255), (0, 0, 0), (127, 0, 0)]
 
            Floats will be rounded down to integers.
 
            Values outside the legal range will be clamped.
 
 
        Will establish a connection to the server as needed.
 
 
        On successful transmission of pixels, return True.
 
        On failure (bad connection), return False.
 
 
        The list of pixel colors will be applied to the LED string starting
 
        with the first LED.  It's not possible to send a color just to one
 
        LED at a time (unless it's the first one).
 
 
        """
 
        self._debug('put_pixels: connecting')
 
        is_connected = self._ensure_connected()
 
        if not is_connected:
 
            self._debug('put_pixels: not connected.  ignoring these pixels.')
 
            return False
 
 
        # build OPC message
 
        len_hi_byte = int(len(pixels)*3 / 256)
 
        len_lo_byte = (len(pixels)*3) % 256
 
        command = 0  # set pixel colors from openpixelcontrol.org
 
 
        header = struct.pack("BBBB", channel, command, len_hi_byte, len_lo_byte)
 
 
        pieces = [ struct.pack( "BBB",
 
                     min(255, max(0, int(r))),
 
                     min(255, max(0, int(g))),
 
                     min(255, max(0, int(b)))) for r, g, b in pixels ]
 
 
        if sys.version_info[0] == 3:
 
            # bytes!
 
            message = header + b''.join(pieces)
 
        else:
 
            # strings!
 
            message = header + ''.join(pieces)
 
 
        self._debug('put_pixels: sending pixels to server')
 
        try:
 
            self._socket.send(message)
 
        except socket.error:
 
            self._debug('put_pixels: connection lost.  could not send pixels.')
 
            self._socket = None
 
            return False
 
 
        if not self._long_connection:
 
            self._debug('put_pixels: disconnecting')
 
            self.disconnect()
 
 
        return True
 
\ No newline at end of file
0 comments (0 inline, 0 general)
Server instance: zserverv-756375 This site is powered by Kallithea, which is © 2010–2021 by various authors & licensed under GPLv3. – Support