st-ten-1/src/components/neo_pixels.py

import sys
import traceback

from PyQt5.QtCore import QThread, pyqtSignal

if "--sim-serial" in sys.argv:
    from components.dummies.serial import serial
else:
    import serial

from ctypes import c_uint32

from .component import Component

C_UINT32_MAX = c_uint32(-1).value


class NeoPixels(Component):
    request = pyqtSignal(object)

    def __init__(self, config=None, name=None):
        self.conn = None
        super().__init__(config=config, name=name, threaded=False)
        self.set_requestors({"self": self.request})

    def config_changed(self):
        self.port = self.config[self.name]["port"]
        self.baudrate = int(self.config[self.name]["baudrate"])
        self.stopbits = getattr(serial, self.config[self.name].get("stopbits", "stopbits_one").upper())
        self.parity = getattr(serial, self.config[self.name].get("parity", "parity_none").upper())
        self.bytesize = getattr(serial, self.config[self.name].get("bytesize", "eightbits").upper())
        self.read_timeout = float(self.config[self.name].get("read_timeout", 1))
        self.write_timeout = float(self.config[self.name].get("write_timeout", 1))
        self.inter_byte_timeout = self.config[self.name].get("inter_byte_timeout", None)
        if self.inter_byte_timeout is not None:
            self.inter_byte_timeout = float(self.inter_byte_timeout)
        if self.conn is not None:
            self.conn.close()
        self.conn = serial.Serial(
            self.port,
            baudrate=self.baudrate,
            stopbits=self.stopbits,
            parity=self.parity,
            bytesize=self.bytesize,
            timeout=self.read_timeout,
            write_timeout=self.write_timeout,
            inter_byte_timeout=self.inter_byte_timeout,
        )
        QThread.msleep(5000)
        self.set_all_pixel_color("#000000")

    @Component.reconfig_on_error
    def _set(self, data=None):
        if not self.ready:
            raise AssertionError("component not ready")
        if not isinstance(data, dict):
            raise ValueError(f"bad call data: {data!r}")
        pixel = data.get("pixel", None)
        color = data.get("color", None)
        if not isinstance(pixel, int) or pixel < 0 or pixel > C_UINT32_MAX:
            raise ValueError(f"the pixel parameter must be a int between 0 and {C_UINT32_MAX} (broadcast)")
        if not isinstance(color, str) or not color.startswith("#") or len(color) != 7:
            raise ValueError("the color parameter must be in the html hex format: '#RRGGBB'")
        try:
            self.conn.write(int.to_bytes(pixel, length=4, byteorder="big") + bytes.fromhex(color[1:7]) + b"\n")
            response = self.conn.readline()
            if response != b"ok\n":
                self.log.error(response.strip())
        except serial.serialutil.SerialException:
            self.log.exception(traceback.format_exc())

    def set(self, pixel, color):
        self.request.emit({"pixel": pixel, "color": color})

    def set_pixel_color(self, pixel, color):
        self.set(pixel, color)

    def set_all_pixel_color(self, color):
        self.set(C_UINT32_MAX, color)