diff options
Diffstat (limited to 'klippy/extras/neopixel.py')
| -rw-r--r-- | klippy/extras/neopixel.py | 71 |
1 files changed, 43 insertions, 28 deletions
diff --git a/klippy/extras/neopixel.py b/klippy/extras/neopixel.py index e72b8a91..1bed21b4 100644 --- a/klippy/extras/neopixel.py +++ b/klippy/extras/neopixel.py @@ -6,27 +6,28 @@ import logging from . import led -BACKGROUND_PRIORITY_CLOCK = 0x7fffffff00000000 +BACKGROUND_PRIORITY_CLOCK = 0x7FFFFFFF00000000 -BIT_MAX_TIME=.000004 -RESET_MIN_TIME=.000050 +BIT_MAX_TIME = 0.000004 +RESET_MIN_TIME = 0.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'] + 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.pin = pin_params["pin"] self.mcu.register_config_callback(self.build_config) self.neopixel_update_cmd = self.neopixel_send_cmd = None # Build color map - chain_count = config.getint('chain_count', 1, minval=1) + chain_count = config.getint("chain_count", 1, minval=1) color_order = config.getlist("color_order", ["GRB"]) if len(color_order) == 1: color_order = [color_order[0]] * chain_count @@ -43,69 +44,83 @@ class PrinterNeoPixel: # Initialize color data self.led_helper = led.LEDHelper(config, self.update_leds, chain_count) self.color_data = bytearray(len(self.color_map)) - self.update_color_data(self.led_helper.get_status()['color_data']) + 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)) + 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) + "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) + "neopixel_send oid=%c", + "neopixel_result oid=%c success=%c", + oid=self.oid, + cq=cmd_queue, + ) + def update_color_data(self, led_state): color_data = self.color_data for cdidx, (lidx, cidx) in self.color_map: - color_data[cdidx] = int(led_state[lidx][cidx] * 255. + .5) + color_data[cdidx] = int(led_state[lidx][cidx] * 255.0 + 0.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] + 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): + for i in range(len(diffs) - 2, -1, -1): pos, count = diffs[i] - nextpos, nextcount = diffs[i+1] + nextpos, nextcount = diffs[i + 1] if pos + 5 >= nextpos and nextcount < 16: diffs[i][1] = nextcount + (nextpos - pos) - del diffs[i+1] + 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) + 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: + 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']: + 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) |