128 lines
4.9 KiB
Python
128 lines
4.9 KiB
Python
"""Optional Yahboom cooling HAT transport; importing it never opens I2C.
|
|
|
|
Built-in effects avoid continuous host writes. Custom RGB remains experimental:
|
|
selecting all LEDs before each channel helped one slow calibration run, but did
|
|
not establish reliable operation across runs or prove any MCU timing guarantee.
|
|
"""
|
|
import time
|
|
import fcntl
|
|
|
|
|
|
class Yahboom:
|
|
def __init__(self, lock, on_error, font):
|
|
from smbus2 import SMBus
|
|
# Ownership belongs to the physical device, not to an API socket path.
|
|
self.ownership = open('/run/lock/pigway-i2c-1-mcu-0d.lock', 'a')
|
|
try:
|
|
fcntl.flock(self.ownership, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
self.bus = SMBus(1)
|
|
except Exception:
|
|
self.ownership.close()
|
|
raise
|
|
self.lock = lock
|
|
self.on_error = on_error
|
|
self.state = None
|
|
self.font = font
|
|
self.fan_level = None
|
|
|
|
def write(self, method, *args):
|
|
with self.lock:
|
|
try:
|
|
return getattr(self.bus, method)(*args)
|
|
except OSError:
|
|
self.state = None
|
|
self.on_error()
|
|
raise
|
|
|
|
def _mcu(self, register, value):
|
|
self.write("write_byte_data", 0x0D, register, value)
|
|
|
|
def off(self):
|
|
with self.lock:
|
|
if self.state == ("off",):
|
|
return
|
|
self._mcu(0x07, 0)
|
|
self.state = ("off",)
|
|
|
|
def effect(self, effect, speed, color, delay_ms=10):
|
|
state = ("effect", effect, speed, color)
|
|
with self.lock:
|
|
if self.state == state:
|
|
return
|
|
# Parameters first, then the documented effect selector. Never
|
|
# restart an unchanged effect when OLED pages or owners change.
|
|
self._mcu(0x05, speed)
|
|
time.sleep(delay_ms / 1000)
|
|
self._mcu(0x06, color)
|
|
time.sleep(delay_ms / 1000)
|
|
self._mcu(0x04, effect)
|
|
self.state = state
|
|
|
|
def color(self, channels):
|
|
channels = tuple(channels)
|
|
if len(channels) != 3 or any(not 0 <= x <= 255 for x in channels):
|
|
raise ValueError("RGB channels must be 0..255")
|
|
state = ("color", channels)
|
|
with self.lock:
|
|
if self.state == state:
|
|
return
|
|
if self.state is None or self.state[0] != "color":
|
|
self._mcu(0x07, 0)
|
|
time.sleep(0.005)
|
|
# One complete RGB update, in fixed R/G/B order. No guessed block
|
|
# write, clear register, or MCU effect command on each flash edge.
|
|
for register, value in enumerate(channels, 1):
|
|
self._mcu(0x00, 0xFF)
|
|
time.sleep(0.005)
|
|
self._mcu(register, value)
|
|
time.sleep(0.005)
|
|
self.state = state
|
|
|
|
def fan(self, level):
|
|
values = (0, 4, 6, 8, 9, 1)
|
|
with self.lock:
|
|
if level != self.fan_level:
|
|
self._mcu(0x08, values[level])
|
|
self.fan_level = level
|
|
|
|
def oled_init(self):
|
|
for c in (0xAE,0xD5,0x80,0xA8,0x1F,0xD3,0x00,0x40,0x8D,0x14,
|
|
0x20,0x00,0xA1,0xC8,0xDA,0x02,0x81,0x7F,0xD9,0xF1,
|
|
0xDB,0x40,0xA4,0xA6,0xAF):
|
|
self.write("write_byte_data", 0x3C, 0, c)
|
|
|
|
def oled(self, lines, home=False):
|
|
# Same 5x7 font and page layout as the verified 128x32 implementation.
|
|
from PIL import Image
|
|
im = Image.new("1", (128,32), 0)
|
|
pix = im.load()
|
|
def text(x,y,value):
|
|
for ch in value:
|
|
code=ord(ch)
|
|
if code*5+5>len(self.font): code=ord("?")
|
|
for cx in range(5):
|
|
col=self.font[code*5+cx]
|
|
for cy in range(7):
|
|
if col & (1<<cy) and 0<=x+cx<128 and 0<=y+cy<32:
|
|
pix[x+cx,y+cy]=1
|
|
x+=6
|
|
if x>122: break
|
|
for row,(left,right) in enumerate(lines[:4]):
|
|
text(0,row*8,str(left))
|
|
if right: text(72 if home else max(0,128-len(str(right))*6),row*8,str(right))
|
|
data=[sum(pix[x,pg*8+bit]<<bit for bit in range(8)) for pg in range(4) for x in range(128)]
|
|
with self.lock:
|
|
for c in (0x21,0,127,0x22,0,3): self.write("write_byte_data",0x3C,0,c)
|
|
for i in range(0,len(data),16): self.write("write_i2c_block_data",0x3C,0x40,data[i:i+16])
|
|
|
|
def close(self):
|
|
errors=[]
|
|
# No fan OFF on shutdown: keep the last cooling level rather than
|
|
# removing cooling merely because the display service is stopping.
|
|
for name,action in (("rgb",self.off),("oled_clear",lambda:self.oled([])),
|
|
("oled_off",lambda:self.write("write_byte_data",0x3C,0,0xAE)),
|
|
("bus",self.bus.close),("ownership",self.ownership.close)):
|
|
try: action()
|
|
except Exception as exc: errors.append((name,str(exc)))
|
|
return errors
|