1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
# 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
from . import led
BACKGROUND_PRIORITY_CLOCK = 0x7FFFFFFF00000000
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"]
self.oid = self.mcu.create_oid()
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)
color_order = config.getlist("color_order", ["GRB"])
if len(color_order) == 1:
color_order = [color_order[0]] * chain_count
if len(color_order) != chain_count:
raise config.error("color_order does not match chain_count")
color_indexes = []
for lidx, co in enumerate(color_order):
if sorted(co) not in (sorted("RGB"), sorted("RGBW")):
raise config.error("Invalid color_order '%s'" % (co,))
color_indexes.extend([(lidx, "RGBW".index(c)) for c in co])
self.color_map = list(enumerate(color_indexes))
if len(self.color_map) > MAX_MCU_SIZE:
raise config.error("neopixel chain too long")
# 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.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_data = self.color_data
for cdidx, (lidx, cidx) in self.color_map:
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]
# 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)
|
