jane/leds
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

reset

Browse source
This commit is contained in:
jane 2021-12-22 21:38:59 -05:00
parent 80729c5fa7
commit 653c132672
15 changed files with 0 additions and 1143 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
config.json
View file

@ -1,9 +0,0 @@
{
"leds": 450,
"brightness": 150,
"gpio": 18,
"type": "grb",
"fade_ticks": 25,
"sleep_time": 100,
"log_level": "INFO"
}

18
index.js
View file

@ -1,18 +0,0 @@
import * as server from './server.js';
import * as lights from './lights.js';
import * as fs from 'fs';
const cfg = JSON.parse(fs.readFileSync('./config.json'));
server.recv(
(res) => {
lights.set_pattern(JSON.parse(res));
},
(res) => {
console.log(`error callback: ${res}`);
}
);
setInterval(() => {
lights.tick();
}, cfg.sleep_time || 500);

235
lcd_manager.py
View file

@ -1,235 +0,0 @@
from time import sleep
class Adafruit_CharLCD:
# commands
LCD_CLEARDISPLAY = 0x01
LCD_RETURNHOME = 0x02
LCD_ENTRYMODESET = 0x04
LCD_DISPLAYCONTROL = 0x08
LCD_CURSORSHIFT = 0x10
LCD_FUNCTIONSET = 0x20
LCD_SETCGRAMADDR = 0x40
LCD_SETDDRAMADDR = 0x80
# flags for display entry mode
LCD_ENTRYRIGHT = 0x00
LCD_ENTRYLEFT = 0x02
LCD_ENTRYSHIFTINCREMENT = 0x01
LCD_ENTRYSHIFTDECREMENT = 0x00
# flags for display on/off control
LCD_DISPLAYON = 0x04
LCD_DISPLAYOFF = 0x00
LCD_CURSORON = 0x02
LCD_CURSOROFF = 0x00
LCD_BLINKON = 0x01
LCD_BLINKOFF = 0x00
# flags for display/cursor shift
LCD_DISPLAYMOVE = 0x08
LCD_CURSORMOVE = 0x00
# flags for display/cursor shift
LCD_DISPLAYMOVE = 0x08
LCD_CURSORMOVE = 0x00
LCD_MOVERIGHT = 0x04
LCD_MOVELEFT = 0x00
# flags for function set
LCD_8BITMODE = 0x10
LCD_4BITMODE = 0x00
LCD_2LINE = 0x08
LCD_1LINE = 0x00
LCD_5x10DOTS = 0x04
LCD_5x8DOTS = 0x00
def __init__(self, pin_rs=24, pin_e=23, pin_b=4, pins_db=[17, 25, 27, 22], GPIO=None):
# Emulate the old behavior of using RPi.GPIO if we haven't been given
# an explicit GPIO interface to use
if not GPIO:
import RPi.GPIO as GPIO
self.GPIO = GPIO
self.pin_rs = pin_rs
self.pin_e = pin_e
self.pin_b = pin_b
self.pins_db = pins_db
self.GPIO.setwarnings(False)
self.GPIO.setmode(GPIO.BCM)
self.GPIO.setup(self.pin_e, GPIO.OUT)
self.GPIO.setup(self.pin_rs, GPIO.OUT)
self.GPIO.setup(self.pin_b, GPIO.OUT)
for pin in self.pins_db:
self.GPIO.setup(pin, GPIO.OUT)
self.GPIO.output(self.pin_b, True)
self.write4bits(0x33) # initialization
self.write4bits(0x32) # initialization
self.write4bits(0x28) # 2 line 5x7 matrix
self.write4bits(0x0C) # turn cursor off 0x0E to enable cursor
self.write4bits(0x06) # shift cursor right
self.displaycontrol = self.LCD_DISPLAYON | self.LCD_CURSOROFF | self.LCD_BLINKOFF
self.displayfunction = self.LCD_4BITMODE | self.LCD_1LINE | self.LCD_5x8DOTS
self.displayfunction |= self.LCD_2LINE
""" Initialize to default text direction (for romance languages) """
self.displaymode = self.LCD_ENTRYLEFT | self.LCD_ENTRYSHIFTDECREMENT
self.write4bits(self.LCD_ENTRYMODESET |
self.displaymode) # set the entry mode
self.clear()
def begin(self, cols, lines):
if (lines > 1):
self.numlines = lines
self.displayfunction |= self.LCD_2LINE
self.currline = 0
def home(self):
self.write4bits(self.LCD_RETURNHOME) # set cursor position to zero
self.delayMicroseconds(3000) # this command takes a long time!
def clear(self):
self.write4bits(self.LCD_CLEARDISPLAY) # command to clear display
# 3000 microsecond sleep, clearing the display takes a long time
self.delayMicroseconds(3000)
def setCursor(self, col, row):
self.row_offsets = [0x00, 0x40, 0x14, 0x54]
if (row > self.numlines):
row = self.numlines - 1 # we count rows starting w/0
self.write4bits(self.LCD_SETDDRAMADDR | (col + self.row_offsets[row]))
def noDisplay(self):
""" Turn the display off (quickly) """
self.GPIO.output(self.pin_b, False)
self.displaycontrol &= ~self.LCD_DISPLAYON
self.write4bits(self.LCD_DISPLAYCONTROL | self.displaycontrol)
def display(self):
""" Turn the display on (quickly) """
self.GPIO.output(self.pin_b, True)
self.displaycontrol |= self.LCD_DISPLAYON
self.write4bits(self.LCD_DISPLAYCONTROL | self.displaycontrol)
def noCursor(self):
""" Turns the underline cursor on/off """
self.displaycontrol &= ~self.LCD_CURSORON
self.write4bits(self.LCD_DISPLAYCONTROL | self.displaycontrol)
def cursor(self):
""" Cursor On """
self.displaycontrol |= self.LCD_CURSORON
self.write4bits(self.LCD_DISPLAYCONTROL | self.displaycontrol)
def noBlink(self):
""" Turn on and off the blinking cursor """
self.displaycontrol &= ~self.LCD_BLINKON
self.write4bits(self.LCD_DISPLAYCONTROL | self.displaycontrol)
def noBlink(self):
""" Turn on and off the blinking cursor """
self.displaycontrol &= ~self.LCD_BLINKON
self.write4bits(self.LCD_DISPLAYCONTROL | self.displaycontrol)
def DisplayLeft(self):
""" These commands scroll the display without changing the RAM """
self.write4bits(self.LCD_CURSORSHIFT |
self.LCD_DISPLAYMOVE | self.LCD_MOVELEFT)
def scrollDisplayRight(self):
""" These commands scroll the display without changing the RAM """
self.write4bits(self.LCD_CURSORSHIFT |
self.LCD_DISPLAYMOVE | self.LCD_MOVERIGHT)
def leftToRight(self):
""" This is for text that flows Left to Right """
self.displaymode |= self.LCD_ENTRYLEFT
self.write4bits(self.LCD_ENTRYMODESET | self.displaymode)
def rightToLeft(self):
""" This is for text that flows Right to Left """
self.displaymode &= ~self.LCD_ENTRYLEFT
self.write4bits(self.LCD_ENTRYMODESET | self.displaymode)
def autoscroll(self):
""" This will 'right justify' text from the cursor """
self.displaymode |= self.LCD_ENTRYSHIFTINCREMENT
self.write4bits(self.LCD_ENTRYMODESET | self.displaymode)
def noAutoscroll(self):
""" This will 'left justify' text from the cursor """
self.displaymode &= ~self.LCD_ENTRYSHIFTINCREMENT
self.write4bits(self.LCD_ENTRYMODESET | self.displaymode)
def write4bits(self, bits, char_mode=False):
""" Send command to LCD """
self.delayMicroseconds(1000) # 1000 microsecond sleep
bits = bin(bits)[2:].zfill(8)
self.GPIO.output(self.pin_rs, char_mode)
for pin in self.pins_db:
self.GPIO.output(pin, False)
for i in range(4):
if bits[i] == "1":
self.GPIO.output(self.pins_db[::-1][i], True)
self.pulseEnable()
for pin in self.pins_db:
self.GPIO.output(pin, False)
for i in range(4, 8):
if bits[i] == "1":
self.GPIO.output(self.pins_db[::-1][i-4], True)
self.pulseEnable()
def delayMicroseconds(self, microseconds):
# divide microseconds by 1 million for seconds
seconds = microseconds / float(1000000)
sleep(seconds)
def pulseEnable(self):
self.GPIO.output(self.pin_e, False)
# 1 microsecond pause - enable pulse must be > 450ns
self.delayMicroseconds(1)
self.GPIO.output(self.pin_e, True)
# 1 microsecond pause - enable pulse must be > 450ns
self.delayMicroseconds(1)
self.GPIO.output(self.pin_e, False)
self.delayMicroseconds(1) # commands need > 37us to settle
def message(self, text):
""" Send string to LCD. Newline wraps to second line"""
for char in text:
if char == '\n':
self.write4bits(0xC0) # next line
else:
self.write4bits(ord(char), True)

100
leds.py
View file

@ -1,100 +0,0 @@
#!/usr/bin/env python
#
# based on code from lrvick and LiquidCrystal
# lrvic - https://github.com/lrvick/raspi-hd44780/blob/master/hd44780.py
# LiquidCrystal - https://github.com/arduino/Arduino/blob/master/libraries/LiquidCrystal/LiquidCrystal.cpp
#
from time import sleep
import sys
import lcd_manager
import light_manager
import pattern
import logger
import threading
import socket
this = sys.modules[__name__]
this.level = 0
this.level_max = 14
this.level_state = 0
this.color_state = 0
def lightlevel(lcd, level):
logger.debug("display level")
lcd.clear()
lcd.message("Light Level:\n]" + "-"*level + "[")
def query():
level_state = 2
def color(lcd):
lcd.clear()
lcd.message("new pattern loaded.")
logger.debug("NYI")
def get_state():
if this.level_state > 0:
this.level_state -= 1
return "level"
elif this.color_state > 0:
this.color_state -= 1
return "color"
else:
return "idle"
def displayon(lcd):
if lcd.displaycontrol & lcd.LCD_DISPLAYON != lcd.displaycontrol:
lcd.display()
def socket_loop():
injected = False
s = socket.create_server(('0.0.0.0', 29999))
while True:
if not injected:
this.lights.set_pattern(pattern.pat)
injected = True
sock, addr = s.accept()
with sock:
length_data = sock.recv(4)
logger.debug(length_data)
length = int(length_data.decode())
logger.debug(length)
pattern_data = sock.recv(length)
logger.debug(pattern_data)
pattern.parse(pattern_data.decode())
this.color_state = 7
def loop():
socket_thread = threading.Thread(target=socket_loop)
lcd = lcd_manager.Adafruit_CharLCD()
this.lights = light_manager.LightStrip(string_length=450, brightness=0.6, max_changes=10)
socket_thread.start()
while True:
logger.debug("loop")
this.lights.tick()
query()
state = get_state()
if state == "level":
if this.lights.get_light_level() != (this.level / this.level_max):
this.lights.set_light_level(this.level / this.level_max)
lightlevel(lcd, this.level)
elif state == "color":
color(lcd)
else:
lcd.noDisplay()
sleep(0.1)
if __name__ == '__main__':
loop()

67
light_manager.py
View file

@ -1,67 +0,0 @@
from time import sleep
import neopixel
import board
import logger
import pattern
def defaultPattern(n, t, pv):
return (t % 255, 255, 255)
class LightStrip:
def __init__(self, data_pin=board.D18, string_length=300, brightness=0.7, pixel_order=neopixel.GRB, max_changes=5):
self.data_pin = data_pin
self.np = neopixel.NeoPixel(
self.data_pin, string_length, brightness=brightness, auto_write=True, pixel_order=pixel_order)
self.pattern = defaultPattern
self.cur_tick = 0
self.cur_index = 0
self.max_changes = max_changes
self.cur_pattern = []
def get_light_level(self):
return self.np.brightness
def set_light_level(self, level):
self.np.brightness = level
def set_pattern(self, pattern_callback):
self.pattern = pattern_callback
def get_next_pattern_tick(self):
np = self.np
n = np.n
t = self.cur_tick
if not len(self.cur_pattern) >= n - 1:
li = [(255, 255, 255)] * (n - len(self.cur_pattern))
self.cur_pattern.extend(li)
for i in range(n):
self.cur_pattern[i] = self.pattern(i, t, np[i])
self.cur_tick = t + 1
def tick(self):
np = self.np
t = self.cur_tick
n = np.n
if not len(self.cur_pattern) >= n - 1:
self.get_next_pattern_tick()
changes = 0
ind = self.cur_index
for i in range(ind, n):
col = self.cur_pattern[i]
self.cur_index = i
logger.debug("TEST VALUES: np {} col {} same {}".format(
tuple(np[i]), col, tuple(np[i]) != col))
if tuple(np[i]) != col:
changes = changes + 1
np[i] = col
logger.debug(
"CHANGE COLOR OF PIXEL {} TO {} ON TICK {}".format(i, col, t))
if changes >= self.max_changes:
break
if self.cur_index >= (n-1):
self.get_next_pattern_tick()
self.cur_index = 0

308
lights.js
View file

@ -1,308 +0,0 @@
import ws281x from "rpi-ws281x";
import * as fs from "fs";
import Logger, { levels } from "./logger.js";
const cfg = JSON.parse(fs.readFileSync("./config.json"));
const log = new Logger(
"lights",
cfg.log_level ? levels[cfg.log_level] : levels.INFO
);
const fade_ticks = cfg.fade_ticks || 20;
var num_ticks = 0;
var pixels = new Uint32Array(cfg.leds);
var pixel_cache = new Uint32Array(cfg.leds);
var next_pattern = new Uint32Array(cfg.leds);
const targets = {};
var pattern = {};
function rgb_to_int(r, g, b) {
let rgb = r;
rgb = (rgb << 8) + g;
rgb = (rgb << 8) + b;
return rgb;
}
function int_to_rgb(int) {
var r = (int >> 16) & 0xff;
var g = (int >> 8) & 0xff;
var b = int & 0xff;
return { r: r, g: g, b: b };
}
ws281x.configure({
leds: cfg.leds || 300,
brightness: cfg.brightness || 200,
gpio: cfg.gpio || 18,
stripType: cfg.type || "grb",
});
export function set_pattern(pat) {
pattern = pat;
log.debug("pattern set");
}
export class Function {
func;
options;
constructor(func, options) {
this.func = func;
this.options = options;
}
}
export const functions = {
random: new Function(
(index, arg1, arg2) => {
return Math.floor(Math.random() * 256);
},
{
requires_arg1: true,
requires_arg2: false,
convert_args: false,
}
),
move: new Function(
(index, arg1, arg2) => {
return arg2;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: true,
}
),
modulo: new Function(
(index, arg1, arg2) => {
return arg1 % arg2;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: true,
}
),
swap: new Function(
(index, arg1, arg2) => {
let temp = targets[arg2];
targets[arg2] = targets[arg1];
return temp;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: false,
}
),
add: new Function(
(index, arg1, arg2) => {
return arg1 + arg2;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: true,
}
),
subtract: new Function(
(index, arg1, arg2) => {
return arg1 - arg2;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: true,
}
),
stagger: new Function(
(index, arg1, arg2) => {
return (index + arg2) % 255;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: true,
}
),
slide: new Function(
(index, arg1, arg2) => {
return (num_ticks + arg2) % 255;
},
{
requires_arg1: true,
requires_arg2: true,
convert_args: true,
}
),
};
function tick_pattern() {
// do the parsing stuff here
log.debug("TICKING PATTERN");
let parsed = pattern.parsed;
log.debug(`${JSON.stringify(pattern)} ${JSON.stringify(parsed)}`);
if (parsed && parsed.length > 0) {
for (let i = 0; i < cfg.leds; i++) {
for (let id in parsed) {
let command = parsed[id];
log.debug(`pattern ${id} ${command}`);
let name = command["command"];
log.debug(`${name} matches: ${functions[name]}`);
if (functions[name] != undefined) {
if (
functions[name].options &&
functions[name].options["convert_args"]
) {
let param_arg1 = parseInt(command.arg1) || targets[command.arg1];
let param_arg2 = parseInt(command.arg2) || targets[command.arg2];
let result = functions[name].func(i, param_arg1, param_arg2);
log.debug(
`convert ${command.arg1} ${param_arg1} ${command.arg2} ${param_arg2} ${result}`
);
targets[command.arg1] = result;
} else {
let result = functions[name].func(i, command.arg1, command.arg2);
log.debug(`no convert ${command.arg1} = ${result}`);
targets[command.arg1] = result;
}
}
}
log.debug(`next: ${targets["r"]}, ${targets["g"]}, ${targets["b"]}`);
next_pattern[i] = rgb_to_int(
targets["r"] || 0,
targets["g"] || 0,
targets["b"] || 0
);
targets["r"] = 0;
targets["g"] = 0;
targets["b"] = 0;
}
} else {
getRandom();
}
num_ticks++;
}
function getRandom() {
for (let i = 0; i < cfg.leds; i++) {
var r = Math.floor(Math.random() * 256);
var g = Math.floor(Math.random() * 256);
var b = Math.floor(Math.random() * 256);
next_pattern[i] = rgb_to_int(r, g, b);
}
}
export function tick() {
var changed = false;
for (let i = 0; i < cfg.leds; i++) {
if (next_pattern[i] != pixels[i]) {
if (next_pattern[i] == pixel_cache[i]) {
log.debug("INCONGRUENCE WITH " + i);
pixels[i] = next_pattern[i];
} else {
changed = true;
fade(i);
}
} else if (pixel_cache[i] != pixels[i]) {
log.debug("PATTERN NOT STORED " + i);
pixel_cache[i] = pixels[i];
}
}
if (!changed) {
tick_pattern();
} else {
ws281x.render(pixels);
}
//ws281x.sleep(cfg.sleep_time || 500);
}
function fade(index) {
var original = int_to_rgb(pixel_cache[index]);
var current = int_to_rgb(pixels[index]);
var final = int_to_rgb(next_pattern[index]);
var diff_r = final.r - original.r;
var diff_cr = current.r - original.r;
var diff_g = final.g - original.g;
var diff_cg = current.g - original.g;
var diff_b = final.b - original.b;
var diff_cb = current.b - original.b;
var sign_r = diff_r === Math.abs(diff_r) ? 1 : -1;
var sign_g = diff_g === Math.abs(diff_g) ? 1 : -1;
var sign_b = diff_b === Math.abs(diff_b) ? 1 : -1;
log.debug(`${diff_r} ${sign_r} ${diff_g} ${sign_g} ${diff_b} ${sign_b}`);
var interval_r = sign_r * Math.ceil(Math.abs(diff_r) / fade_ticks);
var interval_g = sign_g * Math.ceil(Math.abs(diff_g) / fade_ticks);
var interval_b = sign_b * Math.ceil(Math.abs(diff_b) / fade_ticks);
log.debug(
`r${Math.abs(diff_r) / fade_ticks} ${Math.ceil(
Math.abs(diff_r) / fade_ticks
)}` +
`g${Math.abs(diff_g) / fade_ticks} ${Math.ceil(
Math.abs(diff_g) / fade_ticks
)}` +
`b${Math.abs(diff_b) / fade_ticks} ${Math.ceil(
Math.abs(diff_b) / fade_ticks
)}`
);
var current_tick_r = Math.floor(Math.abs(diff_cr / diff_r) * fade_ticks);
var current_tick_g = Math.floor(Math.abs(diff_cg / diff_g) * fade_ticks);
var current_tick_b = Math.floor(Math.abs(diff_cb / diff_b) * fade_ticks);
if (
diff_r == 0 ||
(sign_r == 1 && current.r + interval_r >= final.r) ||
(sign_r == -1 && current.r + interval_r <= final.r) ||
current_tick_r + 1 >= fade_ticks
) {
current.r = final.r;
interval_r = 0;
current_tick_r = fade_ticks;
}
if (
diff_g == 0 ||
(sign_g == 1 && current.g + interval_g >= final.g) ||
(sign_g == -1 && current.g + interval_g <= final.g) ||
current_tick_g + 1 >= fade_ticks
) {
current.g = final.g;
interval_g = 0;
current_tick_g = fade_ticks;
}
if (
diff_b == 0 ||
(sign_b == 1 && current.b + interval_b >= final.b) ||
(sign_b == -1 && current.b + interval_b <= final.b) ||
current_tick_b + 1 >= fade_ticks
) {
current.b = final.b;
interval_b = 0;
current_tick_b = fade_ticks;
}
if (
current_tick_r + 1 >= fade_ticks &&
current_tick_g + 1 >= fade_ticks &&
current_tick_b + 1 >= fade_ticks
) {
log.debug("FINISHED");
pixel_cache[index] = next_pattern[index];
} else {
pixels[index] = rgb_to_int(
current.r + interval_r,
current.g + interval_g,
current.b + interval_b
);
let prev = int_to_rgb(pixel_cache[index]);
log.debug(
`${current_tick_r} ${current_tick_g} ${current_tick_b}: ` +
`CURRENT COLOR: ${current.r} ${current.g} ${current.b} NEW COLOR: ${
current.r + interval_r
} ${current.g + interval_g} ${current.b + interval_b} ` +
`FINAL: ${final.r} ${final.g} ${final.b} ` +
`\nINTERVAL: ${interval_r} ${interval_g} ${interval_b} ` +
`PREVIOUS: ${prev.r} ${prev.g} ${prev.b}`
);
}
}

51
logger.js
View file

@ -1,51 +0,0 @@
export const levels = {
DEBUG: 4,
INFO: 3,
WARN: 2,
ERROR: 1,
PANIC: 0,
};
export default class Logger {
constructor(name, level) {
console.log(`created new logger for ${name} with level ${level}`);
this.sn(name);
this.s(level);
}
n = 'DEFAULT';
l = 0;
sn(n) {
this.n = n;
}
s(l) {
if (l && l.constructor === Number) {
this.l = l;
} else {
this.l = levels[l];
}
}
lo(l, m) {
if (l <= this.l) {
let level = Object.keys(levels).find((key) => levels[key] === l);
let ms = typeof m == 'object' ? JSON.stringify(m) : m;
console.log(`${level} [${this.n}]: ${ms}`);
}
}
debug(msg) {
this.lo(levels.DEBUG, msg);
}
info(msg) {
this.lo(levels.INFO, msg);
}
warn(msg) {
this.lo(levels.WARN, msg);
}
error(msg) {
this.lo(levels.ERROR, msg);
}
panic(msg) {
this.lo(levels.PANIC, msg);
}
}

10
logger.py
View file

@ -1,10 +0,0 @@
debug_statements = False
def debug(msg):
if debug_statements:
print(msg)
def info(msg):
print(msg)

20
package.json
View file

@ -1,20 +0,0 @@
{
"type": "module",
"name": "leds",
"version": "1.0.0",
"description": "",
"main": "index.js",
"scripts": {
"test": "echo \"Error: no test specified\" && exit 1",
"start": "node index.js"
},
"repository": {
"type": "git",
"url": "git@ssh.gitdab.com:jane/leds"
},
"author": "jane@j4.pm",
"license": "UNLICENSED",
"dependencies": {
"rpi-ws281x": "^1.0.36"
}
}

50
parse.js
View file

@ -1,50 +0,0 @@
import * as fs from 'fs';
import Logger, { levels } from './logger.js';
import { functions } from './lights.js';
const cfg = JSON.parse(fs.readFileSync('./config.json'));
const log = new Logger("server", cfg.log_level ? levels[cfg.log_level] : levels.INFO);
export default function parse(data) {
let parsed = []
let errors = []
log.info(data);
// errors.push("not yet implemented");
data = JSON.parse(data);
let line = data.split("\n");
for (let lineNumber in line) {
let currentLine = line[lineNumber];
let split = currentLine.split(" ");
let command = split[0];
let arg1 = split[1];
let arg2 = split[2];
let match = functions[command];
if (match != undefined) {
if (
(match.options["requires_arg1"] && arg1 == undefined) ||
(match.options["requires_arg2"] && arg2 == undefined)
) {
errors.push(`error parsing line ${lineNumber}, invalid number of args`);
}
else if (!match.options["convert_args"] && parseInt(arg1)) {
errors.push(`error parsing line ${lineNumber}, argument ${arg1} cannot be a number`);
}
else {
parsed.push(
{
command: command,
arg1: arg1 || undefined,
arg2: arg2 || undefined
}
);
}
}
}
return {
parsed: parsed,
data: data,
errors: errors
}
}

189
pattern.py
View file

@ -1,189 +0,0 @@
import sys
import json
import random
import logger
this = sys.modules[__name__]
this.encoded = None
this.pattern = None
this.values = {
"stack": 0,
"stack2": 0,
"stack3": 0,
"stack4": 0,
"stack5": 0,
"stack6": 0,
"stackr": 0,
"stackg": 0,
"stackb": 0,
"r": 0,
"g": 0,
"b": 0,
"tick": 0,
"index": 0,
"fadeval": 0,
"fadeinc": True
}
def constant(target, arg, index):
return (arg, index)
def add(target, arg, index):
return (target + arg, index)
def sub(target, arg, index):
return (target - arg, index)
def mult(target, arg, index):
return (target * arg, index)
def div(target, arg, index):
return (target / arg, index)
def mod(target, arg, index):
if arg <= 0:
return target
return (target % arg, index)
def fade(target, arg, index):
value = this.values['fadeval']
if this.values['fadeinc']:
value += 1
if value >= arg:
this.values['fadeinc'] = False
else:
value -= 1
if value <= 0:
this.values['fadeinc'] = True
this.values['fadeval'] = value
return (value, index)
def rand(target, arg, index):
return (random.randrange(0, 255), index)
def jmp(target, arg, index):
return (target, target)
def jnz(target, arg, index):
if target != 0:
return (target, arg)
else:
return (target, index)
def jez(target, arg, index):
if target == 0:
return (target, arg)
else:
return (target, index)
def _apply(index, target, arg, func):
if type(arg) is int:
logger.debug("ran: {} {} {} {}".format(index, func.__name__, target, arg))
return func(target, arg, index)
elif type(arg) is str and arg in this.values:
logger.debug("ran: {} {} {} {}".format(
index, func.__name__, target, this.values[arg]))
return func(target, this.values[arg], index)
def apply(index, targets, args, func):
j = index
for target in range(len(targets)):
if this.values[targets[target]['channel']] != None:
logger.debug("target: {}".format(targets[target]['channel']))
if target < len(args):
val, jump = _apply(
index, this.values[targets[target]['channel']], args[target], func)
if val != this.values[targets[target]['channel']]:
this.values[targets[target]['channel']] = val
j = jump
else:
val, jump = _apply(
index, this.values[targets[target]['channel']], 0, func)
if val != this.values[targets[target]['channel']]:
this.values[targets[target]['channel']] = val
j = jump
return j
this.instructions = {
"CONSTANT": constant,
"ADD": add,
"SUBTRACT": sub,
"MULTIPLY": mult,
"DIVIDE": div,
"MODULO": mod,
"FADE": fade,
"RANDOM": rand,
"JMP": jmp,
"JNZ": jnz,
"JEZ": jez
}
def default(index, tick):
return (((index + tick) * 5) % 255, (tick * 42) % 255, (tick * 50) % 255)
def pat(index, tick, previous_values):
this.values['tick'] = tick
this.values['index'] = index
if this.pattern != None:
i = 0
while i < len(this.pattern):
if i < len(this.pattern):
name = this.pattern[i]['instruction']['name']
targets = this.pattern[i]['instruction']['targets']
args = []
if 'args' in this.pattern[i]['instruction']:
args = this.pattern[i]['instruction']['args']
if this.instructions[name] != None:
jump = apply(i, targets, args, this.instructions[name])
logger.debug("{} {} {} {} {}".format(jump, i, len(
this.pattern), jump != i, jump <= len(this.pattern)))
if jump != i and jump <= len(this.pattern):
logger.debug("jumping to {}".format(jump - 1))
i = jump - 1
i += 1
r = this.values["r"]
g = this.values["g"]
b = this.values["b"]
this.values["r"] = 0
this.values["g"] = 0
this.values["b"] = 0
this.values["stack"] = 0
if r > 255:
r = 255
elif r < 0:
r = 0
if g > 255:
g = 255
elif g < 0:
g = 0
if b > 255:
b = 255
elif b < 0:
b = 0
logger.debug("final color: {}".format((r, g, b)))
return (r, g, b)
else:
return default(index, tick)
def parse(str):
this.encoded = str
logger.debug(this.encoded)
this.pattern = json.loads(this.encoded)

20
pnpm-lock.yaml
View file

@ -1,20 +0,0 @@
lockfileVersion: 5.3
specifiers:
rpi-ws281x: ^1.0.36
dependencies:
rpi-ws281x: 1.0.36
packages:
/nan/2.14.2:
resolution: {integrity: sha512-M2ufzIiINKCuDfBSAUr1vWQ+vuVcA9kqx8JJUsbQi6yf1uGRyb7HfpdfUr5qLXf3B/t8dPvcjhKMmlfnP47EzQ==}
dev: false
/rpi-ws281x/1.0.36:
resolution: {integrity: sha512-iKmc4lKA0/V83q7rdYmFJV3nsQTXaJ8wqyYqdamUn6krDPRevkOhkty9Zj0ZcsYeXBVzxWKzd3VdenDYDQKAGg==}
requiresBuild: true
dependencies:
nan: 2.14.2
dev: false

2
requirements.txt
View file

@ -1,2 +0,0 @@
rpi_ws281x==4.2.5
adafruit-circuitpython-neopixel==6.0.0

2
run.sh
View file

@ -1,2 +0,0 @@
git pull
node index.js

62
server.js
View file

@ -1,62 +0,0 @@
import * as net from 'net';
import * as fs from 'fs';
import Logger, { levels } from './logger.js';
import parse from './parse.js'
const cfg = JSON.parse(fs.readFileSync('./config.json'));
const log = new Logger("server", cfg.log_level ? levels[cfg.log_level] : levels.INFO);
const hostname = '0.0.0.0';
const port = 29999;
export function recv(callback, errorCallback) {
let server = new net.Server();
server.listen(port, hostname, () => {
server.on('connection', (con) => {
console.log('connection recieved: ' +
con.remoteAddress + ":" + con.remotePort);
// let data = [];
// for (let key of Object.keys(functions)) {
// data.push({
// n: key.toUpperCase(),
// a: reqs[key.toLowerCase()] != undefined ? reqs[key.toLowerCase()] : true
// });
// }
// let functions_str = JSON.stringify(data);
// con.write(functions_str);
// log.debug(`sending ${functions_str}`);
con.on('data', (data) => {
let parsed_data = parse(String(data));
if (parsed_data.errors != undefined && parsed_data.errors.length > 0) {
con.write(`error ..\n${JSON.stringify(parsed_data)}`);
errorCallback(JSON.stringify(parsed_data));
}
else {
con.write(`success ..\n${JSON.stringify(parsed_data)}`);
callback(JSON.stringify(parsed_data));
}
});
con.on('close', () => {
console.log('recieved close for ' +
con.remoteAddress + ":" + con.remotePort);
con.destroy();
});
server.getConnections((err, cons) => {
if (err) {
console.error(err);
}
else {
console.log(`connections: ${cons}`);
}
})
})
server.on('error', (e) => {
server.close();
errorCallback(e);
});
});
}