import os, time, threading, colorsys, math, random, datetime from abc import ABC, abstractmethod from typing import List, Tuple from concurrent import futures from dotenv import load_dotenv import grpc import board import neopixel import psutil import lights_pb2 import lights_pb2_grpc Color = Tuple[int, int , int] load_dotenv() def get_env(name) -> str: value = os.getenv(name) if value is None: raise RuntimeError(f"Environment variable not found: {name}") return value def get_env_int(name) -> int: return int(get_env(name)) def get_env_float(name) -> float: return float(get_env(name)) def get_env_str(name) -> str: return get_env(name) def hasattr_all(obj: object, names: List[str]) -> bool: return all([hasattr(obj, name) for name in names]) LED_COUNT = get_env_int("LED_COUNT") LED_PIN_NUM = get_env_int("LED_PIN") LED_PIN = getattr(board, f"D{LED_PIN_NUM}") GAMMA = get_env_float("GAMMA") TIMEOUT_SECONDS = get_env_float("TIMEOUT_SECONDS") IDLE_SPEED = get_env_float("IDLE_SPEED") IDLE_BRIGHTNESS = get_env_float("IDLE_BRIGHTNESS") IDLE_COLOR = ( get_env_int("IDLE_R"), get_env_int("IDLE_G"), get_env_int("IDLE_B") ) class Animation(ABC): @abstractmethod def get_pixels(self, time: float, count: int) -> List[Color]: raise NotImplementedError() class OffAnimation(Animation): def get_pixels(self, time: float, count: int) -> List[Color]: return [(0, 0, 0)] * count class StaticColorAnimation(Animation): def __init__(self, color: Color): self._color = color def get_pixels(self, time: float, count: int) -> List[Color]: return [self._color] * count class StaticGradient2Animation(Animation): def __init__(self, left_color: Color, right_color: Color): self._left_color = left_color self._right_color = right_color def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] if count == 1: return [self._left_color] out = [] (r1, g1, b1) = self._left_color (r2, g2, b2) = self._right_color for i in range(count): pos = i / (count - 1) r = int(r1 * (1 - pos) + r2 * pos) g = int(g1 * (1 - pos) + g2 * pos) b = int(b1 * (1 - pos) + b2 * pos) out.append((r, g, b)) return out class StaticGradient3Animation(Animation): def __init__(self, left_color: Color, middle_color: Color, right_color: Color): self._left_color = left_color self._middle_color = middle_color self._right_color = right_color def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] if count == 1: return [self._left_color] out = [] (c1r, c1g, c1b) = self._left_color (c2r, c2g, c2b) = self._middle_color (c3r, c3g, c3b) = self._right_color for i in range(count): pos = i / (count - 1) if pos < 0.5: f = pos * 2 r = int(c1r * (1 - f) + c2r * f) g = int(c1g * (1 - f) + c2g * f) b = int(c1b * (1 - f) + c2b * f) else: f = (pos - 0.5) * 2 r = int(c2r * (1 - f) + c3r * f) g = int(c2g * (1 - f) + c3g * f) b = int(c2b * (1 - f) + c3b * f) out.append((r, g, b)) return out class StaticRainbowAnimation(Animation): def get_pixels(self, time: float, count: int) -> List[Color]: if count == 0: return [] if count == 1: return [(255, 0, 0)] out = [] for i in range(count): hue = i / count r, g, b = colorsys.hsv_to_rgb(hue, 1.0, 1.0) out.append((int(r * 255), int(g * 255), int(b * 255))) return out class BreathingColorAnimation(Animation): def __init__(self, speed: float, color: Color): self._speed = speed self._color = color def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] intensity = (math.sin(time * self._speed) + 1) / 2 r = int(self._color[0] * intensity) g = int(self._color[1] * intensity) b = int(self._color[2] * intensity) return [(r, g, b)] * count class BreathingGradient2Animation(Animation): def __init__(self, speed: float, left_color: Color, right_color: Color): self._speed = speed self._left_color = left_color self._right_color = right_color def get_pixels(self, time: float, count: int) -> List[Color]: intensity = (math.sin(time * self._speed) + 1) / 2 if count <= 0: return [] if count == 1: return [( int(self._left_color[0] * intensity), int(self._left_color[1] * intensity), int(self._left_color[2] * intensity) )] out = [] (r1, g1, b1) = self._left_color (r2, g2, b2) = self._right_color for i in range(count): pos = i / (count - 1) r = int((r1 * (1 - pos) + r2 * pos) * intensity) g = int((g1 * (1 - pos) + g2 * pos) * intensity) b = int((b1 * (1 - pos) + b2 * pos) * intensity) out.append((r, g, b)) return out class BreathingGradient3Animation(Animation): def __init__(self, speed: float, left_color: Color, middle_color: Color, right_color: Color): self._speed = speed self._left_color = left_color self._middle_color = middle_color self._right_color = right_color def get_pixels(self, time: float, count: int) -> List[Color]: intensity = (math.sin(time * self._speed) + 1) / 2 if count == 0: return [] if count == 1: return [( int(self._left_color[0] * intensity), int(self._left_color[1] * intensity), int(self._left_color[2] * intensity) )] out = [] (c1r, c1g, c1b) = self._left_color (c2r, c2g, c2b) = self._middle_color (c3r, c3g, c3b) = self._right_color for i in range(count): pos = i / (count - 1) if pos < 0.5: f = pos * 2 r = int((c1r * (1 - f) + c2r * f) * intensity) g = int((c1g * (1 - f) + c2g * f) * intensity) b = int((c1b * (1 - f) + c2b * f) * intensity) else: f = (pos - 0.5) * 2 r = int((c2r * (1 - f) + c3r * f) * intensity) g = int((c2g * (1 - f) + c3g * f) * intensity) b = int((c2b * (1 - f) + c3b * f) * intensity) out.append((r, g, b)) return out class BreathingRainbowAnimation(Animation): def __init__(self, speed: float): self._speed = speed def get_pixels(self, time: float, count: int) -> List[Color]: intensity = (math.sin(time * self._speed) + 1) / 2 if count == 0: return [] if count <= 1: return [(255 * intensity, 0, 0)] out = [] for i in range(count): hue = i / count r, g, b = colorsys.hsv_to_rgb(hue, 1.0, 1.0) out.append((int(r * 255 * intensity), int(g * 255 * intensity), int(b * 255 * intensity))) return out class RunningColorAnimation(Animation): def __init__(self, speed: float, length: int, fade: int, color: Color): self._speed = speed self._length = length self._fade = fade self._color = color def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] out = [] center = (time * self._speed) % count for i in range(count): raw_dist = abs(i - center) dist = min(raw_dist, count - raw_dist) if dist <= self._length: intensity = 1.0 elif dist <= self._length + self._fade: fade_pos = dist - self._length intensity = 1.0 - fade_pos / self._fade else: intensity = 0.0 r = int(self._color[0] * intensity) g = int(self._color[1] * intensity) b = int(self._color[2] * intensity) out.append((r, g, b)) return out class RunningGradient2Animation(Animation): def __init__(self, speed: float, left_color: Color, right_color: Color): self._speed = speed self._left_color = left_color self._right_color = right_color def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] (r1, g1, b1) = self._left_color (r2, g2, b2) = self._right_color shift = (time * self._speed) / (count - 1) out = [] for i in range(count): pos = (i / (count - 1) + shift) % 1.0 r = int(r1 * (1 - pos) + r2 * pos) g = int(g1 * (1 - pos) + g2 * pos) b = int(b1 * (1 - pos) + b2 * pos) out.append((r, g, b)) return out class RunningGradient3Animation(Animation): def __init__(self, speed: float, left_color: Color, middle_color: Color, right_color: Color): self._speed = speed self._left_color = left_color self._middle_color = middle_color self._right_color = right_color def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] (c1r, c1g, c1b) = self._left_color (c2r, c2g, c2b) = self._middle_color (c3r, c3g, c3b) = self._right_color shift = (time * self._speed) / (count - 1) out = [] for i in range(count): pos = (i / (count - 1) + shift) % 1.0 if pos < 0.5: f = pos * 2 r = int(c1r * (1 - f) + c2r * f) g = int(c1g * (1 - f) + c2g * f) b = int(c1b * (1 - f) + c2b * f) else: f = (pos - 0.5) * 2 r = int(c2r * (1 - f) + c3r * f) g = int(c2g * (1 - f) + c3g * f) b = int(c2b * (1 - f) + c3b * f) out.append((r, g, b)) return out class RunningRainbowAnimation(Animation): def __init__(self, speed: float): self._speed = speed def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] shift = (time * self._speed) out = [] for i in range(count): hue = (i + shift) % 1.0 r, g, b = colorsys.hsv_to_rgb(hue, 1.0, 1.0) out.append((int(r * 255), int(g * 255), int(b * 255))) return out class SnakeAnimation(Animation): def __init__(self, speed: float, head_length: int, tail_length: int, head_color: Color, tail_color: Color): self._speed = speed self._head_length = head_length self._tail_length = tail_length self._head_color = head_color self._tail_color = tail_color def get_pixels(self, time: float, count: int) -> List[Color]: if count == 0: return [] out = [(0, 0, 0)] * count head_pos = (time * self._speed) % count for i in range(count): dist = (i - head_pos) % count if 0 <= dist < self._head_length: factor = 1.0 elif self._head_length <= dist < self._head_length + self._tail_length: t = dist - self._head_length factor = max(0.0, 1.0 - t / self._tail_length) else: continue (hr, hg, hb) = self._head_color (tr, tg, tb) = self._tail_color r = int(hr * factor + tr * (1 - factor)) g = int(hg * factor + tg * (1 - factor)) b = int(hb * factor + tb * (1 - factor)) out[i] = (r, g, b) return out class FireAnimation(Animation): def __init__(self, speed: float, color: Color): self._speed = speed self._color = color def get_pixels(self, time: float, count: int) -> List[Color]: if count == 0: return [] (r0, g0, b0) = self._color out = [] for i in range(count): flicker = math.sin(i * 0.25 + time * self._speed) flicker = 0.5 + 0.5 * flicker # -> [0..1] flicker *= 0.7 + 0.3 * random.random() r = int(r0 * flicker) g = int(g0 * flicker) b = int(b0 * flicker) out.append((r, g, b)) return out class SnowfallAnimation(Animation): def __init__(self, speed: float, color: Color): self._speed = speed self._color = color self._state = [] def get_pixels(self, time: float, count: int) -> List[Color]: if len(self._state) != count: self._state = [0.0] * count (r0, g0, b0) = self._color out = [] for i in range(count): if self._state[i] <= 0 and random.random() < self._speed * 0.02: self._state[i] = 1.0 self._state[i] = max(0.0, self._state[i] - 0.02 * self._speed) brightness = self._state[i] r = int(r0 * brightness) g = int(g0 * brightness) b = int(b0 * brightness) out.append((r, g, b)) return out class FlashAnimation(Animation): def __init__(self, speed: float, color: Color): self._speed = speed self._color = color self._start_time = None def get_pixels(self, time: float, count: int) -> List[Color]: if count == 0: return [] if self._start_time == None: self._start_time = time (r0, g0, b0) = self._color brightness = math.exp(-(time - self._start_time) * self._speed) brightness = max(0.0, min(1.0, brightness)) r = int(r0 * brightness) g = int(g0 * brightness) b = int(b0 * brightness) return [(r, g, b)] * count class LoadAnimation(Animation): def __init__(self, speed: float, percent: float, last_percent: float, color: Color): self._speed = speed self._percent = max(0.0, min(100.0, percent)) self._last_percent = max(0.0, min(100.0, last_percent)) self._color = color self._last_time = None def get_pixels(self, time: float, count: int) -> List[Color]: if count <= 0: return [] if self._last_time == None: self._last_time = time dt = time - self._last_time self._last_time = time smoothed_percent = self._last_percent + (self._percent - self._last_percent) * min(1.0, self._speed * dt) (r0, g0, b0) = self._color filled_float = (smoothed_percent / 100.0) * count filled = int(filled_float) fractional = filled_float - filled out = [] for i in range(count): if i < filled: out.append((r0, g0, b0)) elif i == filled and fractional > 0: r = int(r0 * fractional) g = int(g0 * fractional) b = int(b0 * fractional) out.append((r, g, b)) else: out.append((0, 0, 0)) # фиксируем новое сглаженное значение как last_percent self._last_percent = smoothed_percent return out class CpuLoadAnimation(LoadAnimation): def __init__(self, speed: float, color: Color): super().__init__(speed, 0.0, 0.0, color) def get_pixels(self, time: float, count: int) -> List[Color]: self._percent = psutil.cpu_percent(interval=0.0) pixels = super().get_pixels(time, count) return pixels class IdleAnimationUsual(RunningColorAnimation): def __init__(self): super().__init__(IDLE_SPEED, 1, 4, IDLE_COLOR) def get_pixels(self, time: float, count: int) -> List[Color]: return super().get_pixels(time, count) class IdleAnimationNewYear(RunningRainbowAnimation): # Новогодняя пасхалка def __init__(self): super().__init__(IDLE_SPEED / 10) def get_pixels(self, time: float, count: int) -> List[Color]: return super().get_pixels(time, count) today = datetime.datetime.now() if (today.month == 12 and today.day == 31) or (today.month == 1 and today.day == 1): IdleAnimation = IdleAnimationNewYear else: IdleAnimation = IdleAnimationUsual class LightsDaemon(lights_pb2_grpc.LightsServicer): def __init__(self): self._running = True self._lock = threading.Lock() with self._lock: self._pixels = neopixel.NeoPixel( LED_PIN, LED_COUNT, brightness=IDLE_BRIGHTNESS, auto_write=False ) self._gamma = [int((i / 255) ** GAMMA * 255 + 0.5) for i in range(256)] self._color_clamp = lambda x: max(0, min(255, x)) self._control_mode = False self._last_ping = 0 self._animation = IdleAnimation() self._worker = threading.Thread(target=self.loop, daemon=True) self._worker.start() def stop(self): self._running = False self._worker.join() for i in range(len(self._pixels)): self._pixels[i] = (0, 0, 0) self._pixels.show() def loop(self): while self._running: with self._lock: if self._control_mode and time.time() - self._last_ping > TIMEOUT_SECONDS: self._control_mode = False self._pixels.brightness = IDLE_BRIGHTNESS self._animation = IdleAnimation() self._run_animation() time.sleep(0.02) def _run_animation(self): new_colors = self._animation.get_pixels(time.time(), LED_COUNT) for i in range(LED_COUNT): self._pixels[i] = new_colors[i] self._pixels.show() def _update_last_ping(self): with self._lock: self._control_mode = True self._last_ping = time.time() def Disconnect(self, request, context): with self._lock: self._control_mode = False self._pixels.brightness = IDLE_BRIGHTNESS self._animation = IdleAnimation() return lights_pb2.Empty() def Ping(self, request, context): self._update_last_ping() return lights_pb2.Empty() def SetOff(self, request, context): self._update_last_ping() with self._lock: self._control_mode = True self._animation = OffAnimation() return lights_pb2.Empty() def SetStaticColor(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = StaticColorAnimation(color) return lights_pb2.Empty() def SetStaticGradient2(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "left_r", "left_g", "left_b", "right_r", "right_g", "right_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness left_color = ( self._gamma[self._color_clamp(request.left_r)], self._gamma[self._color_clamp(request.left_g)], self._gamma[self._color_clamp(request.left_b)], ) right_color = ( self._gamma[self._color_clamp(request.right_r)], self._gamma[self._color_clamp(request.right_g)], self._gamma[self._color_clamp(request.right_b)], ) self._animation = StaticGradient2Animation(left_color, right_color) return lights_pb2.Empty() def SetStaticGradient3(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "left_r", "left_g", "left_b", "middle_r", "middle_g", "middle_b", "right_r", "right_g", "right_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness left_color = ( self._gamma[self._color_clamp(request.left_r)], self._gamma[self._color_clamp(request.left_g)], self._gamma[self._color_clamp(request.left_b)], ) middle_color = ( self._gamma[self._color_clamp(request.middle_r)], self._gamma[self._color_clamp(request.middle_g)], self._gamma[self._color_clamp(request.middle_b)], ) right_color = ( self._gamma[self._color_clamp(request.right_r)], self._gamma[self._color_clamp(request.right_g)], self._gamma[self._color_clamp(request.right_b)], ) self._animation = StaticGradient3Animation(left_color, middle_color, right_color) return lights_pb2.Empty() def SetStaticRainbow(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness self._animation = StaticRainbowAnimation() return lights_pb2.Empty() def SetBreathingColor(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = BreathingColorAnimation(request.speed, color) return lights_pb2.Empty() def SetBreathingGradient2(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "left_r", "left_g", "left_b", "right_r", "right_g", "right_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness left_color = ( self._gamma[self._color_clamp(request.left_r)], self._gamma[self._color_clamp(request.left_g)], self._gamma[self._color_clamp(request.left_b)], ) right_color = ( self._gamma[self._color_clamp(request.right_r)], self._gamma[self._color_clamp(request.right_g)], self._gamma[self._color_clamp(request.right_b)], ) self._animation = BreathingGradient2Animation(request.speed, left_color, right_color) return lights_pb2.Empty() def SetBreathingGradient3(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "left_r", "left_g", "left_b", "middle_r", "middle_g", "middle_b", "right_r", "right_g", "right_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness left_color = ( self._gamma[self._color_clamp(request.left_r)], self._gamma[self._color_clamp(request.left_g)], self._gamma[self._color_clamp(request.left_b)], ) middle_color = ( self._gamma[self._color_clamp(request.middle_r)], self._gamma[self._color_clamp(request.middle_g)], self._gamma[self._color_clamp(request.middle_b)], ) right_color = ( self._gamma[self._color_clamp(request.right_r)], self._gamma[self._color_clamp(request.right_g)], self._gamma[self._color_clamp(request.right_b)], ) self._animation = BreathingGradient3Animation(request.speed, left_color, middle_color, right_color) return lights_pb2.Empty() def SetBreathingRainbow(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness self._animation = BreathingRainbowAnimation(request.speed) return lights_pb2.Empty() def SetRunningColor(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "length", "fade", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = RunningColorAnimation(request.speed, request.length, request.fade, color) return lights_pb2.Empty() def SetRunningGradient2(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "left_r", "left_g", "left_b", "right_r", "right_g", "right_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness left_color = ( self._gamma[self._color_clamp(request.left_r)], self._gamma[self._color_clamp(request.left_g)], self._gamma[self._color_clamp(request.left_b)], ) right_color = ( self._gamma[self._color_clamp(request.right_r)], self._gamma[self._color_clamp(request.right_g)], self._gamma[self._color_clamp(request.right_b)], ) self._animation = RunningGradient2Animation(request.speed, left_color, right_color) return lights_pb2.Empty() def SetRunningGradient3(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "left_r", "left_g", "left_b", "middle_r", "middle_g", "middle_b", "right_r", "right_g", "right_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness left_color = ( self._gamma[self._color_clamp(request.left_r)], self._gamma[self._color_clamp(request.left_g)], self._gamma[self._color_clamp(request.left_b)], ) middle_color = ( self._gamma[self._color_clamp(request.middle_r)], self._gamma[self._color_clamp(request.middle_g)], self._gamma[self._color_clamp(request.middle_b)], ) right_color = ( self._gamma[self._color_clamp(request.right_r)], self._gamma[self._color_clamp(request.right_g)], self._gamma[self._color_clamp(request.right_b)], ) self._animation = RunningGradient3Animation(request.speed, left_color, middle_color, right_color) return lights_pb2.Empty() def SetRunningRainbow(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness self._animation = RunningRainbowAnimation(request.speed) return lights_pb2.Empty() def SetSnake(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "head_length", "tail_length", "head_r", "head_g", "head_b", "tail_r", "tail_g", "tail_b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness head_color = ( self._gamma[self._color_clamp(request.head_r)], self._gamma[self._color_clamp(request.head_g)], self._gamma[self._color_clamp(request.head_b)], ) tail_color = ( self._gamma[self._color_clamp(request.tail_r)], self._gamma[self._color_clamp(request.tail_g)], self._gamma[self._color_clamp(request.tail_b)], ) self._animation = SnakeAnimation(request.speed, request.head_length, request.tail_length, head_color, tail_color) return lights_pb2.Empty() def SetFire(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = FireAnimation(request.speed, color) return lights_pb2.Empty() def SetSnowfall(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = SnowfallAnimation(request.speed, color) return lights_pb2.Empty() def SetFlash(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = FlashAnimation(request.speed, color) return lights_pb2.Empty() def SetLoad(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "percent", "last_percent", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = LoadAnimation(request.speed, request.percent, request.last_percent, color) return lights_pb2.Empty() def SetCpuLoad(self, request, context): self._update_last_ping() if not hasattr_all(request, ["brightness", "speed", "r", "g", "b"]): return lights_pb2.Empty() with self._lock: self._control_mode = True self._pixels.brightness = request.brightness color = ( self._gamma[self._color_clamp(request.r)], self._gamma[self._color_clamp(request.g)], self._gamma[self._color_clamp(request.b)], ) self._animation = CpuLoadAnimation(request.speed, color) return lights_pb2.Empty() def serve(): daemon = LightsDaemon() server = grpc.server(futures.ThreadPoolExecutor(max_workers=1)) lights_pb2_grpc.add_LightsServicer_to_server(daemon, server) server.add_insecure_port('localhost:50051') server.start() print("Lights daemon started on port 50051") try: server.wait_for_termination() except KeyboardInterrupt: print("Shutting down") server.stop(0) daemon.stop() print("Lights daemon stopped") if __name__ == "__main__": serve()