Artem Kashaev
30feef708c
- Adjust angle parameters for switches in config.json - Implement semaphore loading in main.py - Refactor Lamp modes to use integers in railway_signal.py - Add set_mode method for Lamp class to control lamp states
328 lines
9.5 KiB
Python
328 lines
9.5 KiB
Python
import machine
|
||
import neopixel
|
||
import time
|
||
|
||
try:
|
||
import micropython
|
||
except ImportError: # CPython / non-MicroPython
|
||
micropython = None # type: ignore
|
||
|
||
LAMP_COLORS = {
|
||
# Имена цветов -> RGBA (для RGBW ленты)
|
||
"OFF": (0, 0, 0, 0),
|
||
"RED": (100, 0, 0, 0),
|
||
"YELLOW": (100, 45, 0, 0),
|
||
"GREEN": (0, 100, 0, 0),
|
||
"BLUE": (0, 0, 100, 0),
|
||
"WHITE": (0, 0, 0, 100),
|
||
"MOON_WHITE": (0, 0, 0, 100),
|
||
}
|
||
|
||
|
||
class Lamp:
|
||
"""Одна лампа светофора (один пиксель).
|
||
|
||
3 режима:
|
||
- off: выключено
|
||
- on: постоянно горит
|
||
- blink: мигает (переключение ON/OFF с периодом period_ms)
|
||
"""
|
||
|
||
ERROR = 1
|
||
MODE_OFF = 2
|
||
MODE_ON = 3
|
||
MODE_BLINK = 4
|
||
|
||
def __init__(self, signal: "RailwaySignal", lamp_id: int, pixel_index: int, rgba):
|
||
self._signal = signal
|
||
self.id = lamp_id
|
||
self.pixel = pixel_index
|
||
self.rgba = rgba
|
||
|
||
self.mode = Lamp.MODE_OFF
|
||
self._blink_period_ms = 500
|
||
self._blink_on = False
|
||
self._next_toggle_ms = 0
|
||
|
||
def off(self):
|
||
self._signal._set_lamp_mode(self, Lamp.MODE_OFF)
|
||
|
||
def on(self):
|
||
self._signal._set_lamp_mode(self, Lamp.MODE_ON)
|
||
|
||
def blink(self, period_ms: int = 500):
|
||
self._signal._set_lamp_mode(self, Lamp.MODE_BLINK, period_ms=period_ms)
|
||
|
||
def set_mode(self, mode: int, *, period_ms: int = 500):
|
||
self._signal._set_lamp_mode(self, mode, period_ms=period_ms)
|
||
|
||
|
||
class RailwaySignal:
|
||
"""Контейнер ламп светофора.
|
||
|
||
Хранит NeoPixel и набор ламп (каждая лампа — один пиксель).
|
||
Мигающие лампы обслуживаются общим Timer.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
pin: int,
|
||
num_leds: int,
|
||
*,
|
||
bpp: int = 4,
|
||
timing: int = 1,
|
||
colors: dict | None = None,
|
||
timer_id: int = -1,
|
||
timer_tick_ms: int = 50,
|
||
brightness: float = 0.5,
|
||
):
|
||
self.pin = pin
|
||
self.num_leds = num_leds
|
||
self.bpp = bpp
|
||
self.timing = timing
|
||
|
||
self.colors = colors if colors is not None else LAMP_COLORS
|
||
|
||
self.brightness = float(brightness)
|
||
if self.brightness < 0.0:
|
||
self.brightness = 0.0
|
||
if self.brightness > 1.0:
|
||
self.brightness = 1.0
|
||
|
||
self._timer_id = timer_id
|
||
self._timer_tick_ms = int(timer_tick_ms)
|
||
self._timer = None
|
||
self._timer_running = False
|
||
self._scheduled = False
|
||
|
||
self._lamps_by_id = {}
|
||
|
||
self._np = neopixel.NeoPixel(
|
||
machine.Pin(pin),
|
||
num_leds,
|
||
bpp=bpp,
|
||
timing=timing,
|
||
)
|
||
|
||
self.clear()
|
||
|
||
def set_brightness(self, brightness: float):
|
||
"""Устанавливает глобальный коэффициент яркости (0.0..1.0)."""
|
||
|
||
self.brightness = float(brightness)
|
||
if self.brightness < 0.0:
|
||
self.brightness = 0.0
|
||
if self.brightness > 1.0:
|
||
self.brightness = 1.0
|
||
|
||
# Перерисовываем текущее состояние ламп
|
||
for lamp in self._lamps_by_id.values():
|
||
self._apply_lamp(lamp)
|
||
self.show()
|
||
|
||
def _scale(self, value: int) -> int:
|
||
value = int(value)
|
||
if value <= 0:
|
||
return 0
|
||
scaled = int(value * self.brightness)
|
||
if scaled < 0:
|
||
return 0
|
||
if scaled > 255:
|
||
return 255
|
||
return scaled
|
||
|
||
def _normalize_rgba(self, rgba: tuple | list):
|
||
"""Приводит цвет к формату (r,g,b,w) с учётом self.bpp.
|
||
|
||
- для bpp=4: всегда (r,g,b,w)
|
||
- для bpp=3: W-канал маппится в RGB только если RGB=0 (белый)
|
||
"""
|
||
|
||
if len(rgba) == 3:
|
||
r, g, b = rgba
|
||
w = 0
|
||
else:
|
||
r, g, b, w = rgba
|
||
|
||
r = int(r)
|
||
g = int(g)
|
||
b = int(b)
|
||
w = int(w)
|
||
|
||
if self.bpp == 3:
|
||
# На RGB-ленте нет W-канала. Самый ожидаемый кейс — "WHITE"/"MOON_WHITE",
|
||
# которые заданы как (0,0,0,w). Маппим их в (w,w,w).
|
||
if r == 0 and g == 0 and b == 0 and w > 0:
|
||
r = w
|
||
g = w
|
||
b = w
|
||
w = 0
|
||
|
||
return (r, g, b, w)
|
||
|
||
def add_lamp(self, lamp_id: int, pixel_index: int, color: str | tuple | list):
|
||
if pixel_index < 0 or pixel_index >= self.num_leds:
|
||
raise ValueError("pixel_index out of range")
|
||
|
||
rgba = None
|
||
if isinstance(color, str):
|
||
rgba = self.colors.get(color.upper())
|
||
elif isinstance(color, (tuple, list)):
|
||
rgba = color
|
||
|
||
if rgba is None:
|
||
raise ValueError("Unknown color")
|
||
|
||
rgba = self._normalize_rgba(rgba)
|
||
|
||
lamp = Lamp(self, lamp_id, pixel_index, rgba)
|
||
self._lamps_by_id[lamp_id] = lamp
|
||
lamp.off()
|
||
return lamp
|
||
|
||
def lamp(self, lamp_id: int):
|
||
return self._lamps_by_id.get(lamp_id)
|
||
|
||
def _stop_timer(self):
|
||
if self._timer is None or not self._timer_running:
|
||
return
|
||
try:
|
||
self._timer.deinit()
|
||
except Exception:
|
||
pass
|
||
self._timer_running = False
|
||
|
||
def _ensure_timer(self):
|
||
if self._timer_running:
|
||
return
|
||
|
||
try:
|
||
self._timer = machine.Timer(self._timer_id)
|
||
except Exception:
|
||
# Some ports require Timer(0/1/2...), some allow -1.
|
||
self._timer = machine.Timer(0)
|
||
|
||
try:
|
||
self._timer.init(period=self._timer_tick_ms, mode=machine.Timer.PERIODIC, callback=self._timer_cb)
|
||
self._timer_running = True
|
||
except Exception:
|
||
self._timer_running = False
|
||
|
||
def _now_ms(self) -> int:
|
||
ticks_ms = getattr(time, "ticks_ms", None)
|
||
if ticks_ms is not None:
|
||
return int(ticks_ms())
|
||
return int(time.time() * 1000)
|
||
|
||
def _ticks_diff(self, a: int, b: int) -> int:
|
||
ticks_diff = getattr(time, "ticks_diff", None)
|
||
if ticks_diff is not None:
|
||
return int(ticks_diff(a, b))
|
||
return int(a - b)
|
||
|
||
def show(self):
|
||
self._np.write()
|
||
|
||
def clear(self):
|
||
self._stop_timer()
|
||
self._scheduled = False
|
||
if self.bpp == 3:
|
||
self._np.fill((0, 0, 0))
|
||
else:
|
||
self._np.fill((0, 0, 0, 0))
|
||
self.show()
|
||
|
||
def set_pixel_rgba(self, index: int, r: int, g: int, b: int, w: int = 0):
|
||
if self.bpp == 3:
|
||
self._np[index] = (self._scale(r), self._scale(g), self._scale(b))
|
||
else:
|
||
self._np[index] = (
|
||
self._scale(r),
|
||
self._scale(g),
|
||
self._scale(b),
|
||
self._scale(w),
|
||
)
|
||
|
||
def fill_rgba(self, r: int, g: int, b: int, w: int = 0):
|
||
if self.bpp == 3:
|
||
self._np.fill((self._scale(r), self._scale(g), self._scale(b)))
|
||
else:
|
||
self._np.fill((self._scale(r), self._scale(g), self._scale(b), self._scale(w)))
|
||
|
||
def _apply_lamp(self, lamp: Lamp):
|
||
if lamp.mode == Lamp.MODE_ON:
|
||
r, g, b, w = lamp.rgba
|
||
self.set_pixel_rgba(lamp.pixel, r, g, b, w)
|
||
elif lamp.mode == Lamp.MODE_BLINK:
|
||
if lamp._blink_on:
|
||
r, g, b, w = lamp.rgba
|
||
self.set_pixel_rgba(lamp.pixel, r, g, b, w)
|
||
else:
|
||
self.set_pixel_rgba(lamp.pixel, 0, 0, 0, 0)
|
||
else:
|
||
self.set_pixel_rgba(lamp.pixel, 0, 0, 0, 0)
|
||
|
||
def _any_blinking(self) -> bool:
|
||
for lamp in self._lamps_by_id.values():
|
||
if lamp.mode == Lamp.MODE_BLINK:
|
||
return True
|
||
return False
|
||
|
||
def _set_lamp_mode(self, lamp: Lamp, mode: int, *, period_ms: int = 500):
|
||
if mode == Lamp.MODE_BLINK:
|
||
lamp.mode = Lamp.MODE_BLINK
|
||
lamp._blink_period_ms = int(period_ms)
|
||
lamp._blink_on = False # стартуем с OFF как раньше
|
||
lamp._next_toggle_ms = self._now_ms() + lamp._blink_period_ms
|
||
elif mode == Lamp.MODE_ON:
|
||
lamp.mode = Lamp.MODE_ON
|
||
lamp._blink_on = False
|
||
else:
|
||
lamp.mode = Lamp.MODE_OFF
|
||
lamp._blink_on = False
|
||
|
||
self._apply_lamp(lamp)
|
||
self.show()
|
||
|
||
if self._any_blinking():
|
||
self._ensure_timer()
|
||
else:
|
||
self._stop_timer()
|
||
|
||
def _timer_cb(self, _t):
|
||
# IRQ context: keep it tiny.
|
||
if not self._any_blinking():
|
||
return
|
||
|
||
if micropython is not None:
|
||
if self._scheduled:
|
||
return
|
||
try:
|
||
self._scheduled = True
|
||
micropython.schedule(self._scheduled_tick, 0)
|
||
except Exception:
|
||
self._scheduled = False
|
||
|
||
def _scheduled_tick(self, _arg):
|
||
self._scheduled = False
|
||
|
||
now_ms = self._now_ms()
|
||
dirty = False
|
||
|
||
for lamp in self._lamps_by_id.values():
|
||
if lamp.mode != Lamp.MODE_BLINK:
|
||
continue
|
||
if self._ticks_diff(now_ms, lamp._next_toggle_ms) < 0:
|
||
continue
|
||
|
||
lamp._blink_on = not lamp._blink_on
|
||
lamp._next_toggle_ms = now_ms + lamp._blink_period_ms
|
||
self._apply_lamp(lamp)
|
||
dirty = True
|
||
|
||
if dirty:
|
||
self.show()
|
||
|
||
if not self._any_blinking():
|
||
self._stop_timer()
|