mirror of https://github.com/Desuuuu/klipper.git
112 lines
4.9 KiB
Python
112 lines
4.9 KiB
Python
# Support for "neopixel" leds
|
|
#
|
|
# Copyright (C) 2019-2022 Kevin O'Connor <kevin@koconnor.net>
|
|
#
|
|
# This file may be distributed under the terms of the GNU GPLv3 license.
|
|
import logging
|
|
|
|
BACKGROUND_PRIORITY_CLOCK = 0x7fffffff00000000
|
|
|
|
BIT_MAX_TIME=.000004
|
|
RESET_MIN_TIME=.000050
|
|
|
|
MAX_MCU_SIZE = 500 # Sanity check on LED chain length
|
|
|
|
class PrinterNeoPixel:
|
|
def __init__(self, config):
|
|
self.printer = printer = config.get_printer()
|
|
self.mutex = printer.get_reactor().mutex()
|
|
# Configure neopixel
|
|
ppins = printer.lookup_object('pins')
|
|
pin_params = ppins.lookup_pin(config.get('pin'))
|
|
self.mcu = pin_params['chip']
|
|
self.oid = self.mcu.create_oid()
|
|
self.pin = pin_params['pin']
|
|
self.mcu.register_config_callback(self.build_config)
|
|
color_order = config.get("color_order", "GRB")
|
|
if sorted(color_order) not in (sorted("RGB"), sorted("RGBW")):
|
|
raise config.error("Invalid color_order '%s'" % (color_order,))
|
|
if 'W' in color_order:
|
|
color_index = [color_order.index(c) for c in "RGBW"]
|
|
else:
|
|
color_index = [color_order.index(c) for c in "RGB"]
|
|
self.color_map = list(enumerate(color_index))
|
|
elem_size = len(self.color_map)
|
|
self.chain_count = config.getint('chain_count', 1, minval=1,
|
|
maxval=MAX_MCU_SIZE//elem_size)
|
|
self.neopixel_update_cmd = self.neopixel_send_cmd = None
|
|
# Initialize color data
|
|
pled = printer.load_object(config, "led")
|
|
self.led_helper = pled.setup_helper(config, self.update_leds,
|
|
self.chain_count)
|
|
self.color_data = bytearray(self.chain_count * elem_size)
|
|
self.update_color_data(self.led_helper.get_status()['color_data'])
|
|
self.old_color_data = bytearray([d ^ 1 for d in self.color_data])
|
|
# Register callbacks
|
|
printer.register_event_handler("klippy:connect", self.send_data)
|
|
def build_config(self):
|
|
bmt = self.mcu.seconds_to_clock(BIT_MAX_TIME)
|
|
rmt = self.mcu.seconds_to_clock(RESET_MIN_TIME)
|
|
self.mcu.add_config_cmd("config_neopixel oid=%d pin=%s data_size=%d"
|
|
" bit_max_ticks=%d reset_min_ticks=%d"
|
|
% (self.oid, self.pin, len(self.color_data),
|
|
bmt, rmt))
|
|
cmd_queue = self.mcu.alloc_command_queue()
|
|
self.neopixel_update_cmd = self.mcu.lookup_command(
|
|
"neopixel_update oid=%c pos=%hu data=%*s", cq=cmd_queue)
|
|
self.neopixel_send_cmd = self.mcu.lookup_query_command(
|
|
"neopixel_send oid=%c", "neopixel_result oid=%c success=%c",
|
|
oid=self.oid, cq=cmd_queue)
|
|
def update_color_data(self, led_state):
|
|
color_map = self.color_map
|
|
elem_size = len(color_map)
|
|
color_data = self.color_data
|
|
for i, color in enumerate(led_state):
|
|
for lidx, cidx in color_map:
|
|
color_data[i * elem_size + cidx] = int(color[lidx] * 255. + .5)
|
|
def send_data(self, print_time=None):
|
|
old_data, new_data = self.old_color_data, self.color_data
|
|
if new_data == old_data:
|
|
return
|
|
# Find the position of all changed bytes in this framebuffer
|
|
diffs = [[i, 1] for i, (n, o) in enumerate(zip(new_data, old_data))
|
|
if n != o]
|
|
# Batch together changes that are close to each other
|
|
for i in range(len(diffs)-2, -1, -1):
|
|
pos, count = diffs[i]
|
|
nextpos, nextcount = diffs[i+1]
|
|
if pos + 5 >= nextpos and nextcount < 16:
|
|
diffs[i][1] = nextcount + (nextpos - pos)
|
|
del diffs[i+1]
|
|
# Transmit changes
|
|
ucmd = self.neopixel_update_cmd.send
|
|
for pos, count in diffs:
|
|
ucmd([self.oid, pos, new_data[pos:pos+count]],
|
|
reqclock=BACKGROUND_PRIORITY_CLOCK)
|
|
old_data[:] = new_data
|
|
# Instruct mcu to update the LEDs
|
|
minclock = 0
|
|
if print_time is not None:
|
|
minclock = self.mcu.print_time_to_clock(print_time)
|
|
scmd = self.neopixel_send_cmd.send
|
|
if self.printer.get_start_args().get('debugoutput') is not None:
|
|
return
|
|
for i in range(8):
|
|
params = scmd([self.oid], minclock=minclock,
|
|
reqclock=BACKGROUND_PRIORITY_CLOCK)
|
|
if params['success']:
|
|
break
|
|
else:
|
|
logging.info("Neopixel update did not succeed")
|
|
def update_leds(self, led_state, print_time):
|
|
def reactor_bgfunc(eventtime):
|
|
with self.mutex:
|
|
self.update_color_data(led_state)
|
|
self.send_data(print_time)
|
|
self.printer.get_reactor().register_callback(reactor_bgfunc)
|
|
def get_status(self, eventtime=None):
|
|
return self.led_helper.get_status(eventtime)
|
|
|
|
def load_config_prefix(config):
|
|
return PrinterNeoPixel(config)
|