forked from amkovkov/GranuSightSoftware2
933 lines
32 KiB
Python
933 lines
32 KiB
Python
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 = get_env_float("TIMEOUT")
|
|
|
|
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:
|
|
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 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()
|