1"""RGB LED indicator wrapper for NeoPixel status display.
2
3Drives a NeoPixel LED strip to indicate device status (network connectivity,
4data sending, errors) through color and blink/fade patterns, and optionally
5shows AQI-based beacon colors on remaining LEDs. Pixel ``0`` carries the
6status color/pattern produced by the :class:`Magic` animation engine; pixels
7``1..N-1`` carry the beacon color resolved from the live AQI queue.
8
9The module also exposes lightweight helpers :func:`hex_to_rgb` and
10:func:`get_color_from_value` for converting hex strings and mapping AQI
11values to beacon colors, plus a :func:`main` event loop that the parent
12process spawns as a long-running indicator worker.
13
14Example:
15 Run the indicator loop with a network status queue and beacon queue::
16
17 >>> from queue import Queue # doctest: +SKIP
18 >>> from OzWrapper.OzRGB import OzRGB # doctest: +SKIP
19 >>> net_q, beacon_q = Queue(), Queue() # doctest: +SKIP
20 >>> net_q.put(1) # 1 -> Connected (cyan, fade) # doctest: +SKIP
21 >>> OzRGB.main(None, net_q, beacon_q) # doctest: +SKIP
22
23Note:
24 The NeoPixel data line is driven via the on-board PWM peripheral on
25 GPIO18 (``board.D18``); the NeoPixel WS281x protocol requires an
26 800 kHz signal with sub-microsecond timing, so the pin and frequency
27 must not be reconfigured at runtime. Ten LEDs are addressed by default
28 (``num_pixels = 10``); index ``0`` is reserved for status and indices
29 ``1..9`` form the AQI beacon ring.
30
31Reference config:
32
33.. code-block:: json
34
35 {
36 "beacon": {
37 "en": 1,
38 "sc": "aqi",
39 "range": [
40 0,
41 50,
42 100,
43 200,
44 300,
45 400
46 ],
47 "colors": [
48 "50f0e6",
49 "29cc70",
50 "f0e641",
51 "d08000",
52 "ff0000",
53 "c700d0"
54 ]
55 }
56 }
57
58See Also:
59 :class:`SensorBase.SensorBase.GenericSensor`: Common sensor wrapper
60 base class used across the Oizom hardware peripherals.
61"""
62
63import time
64from queue import Queue
65
66import board
67import neopixel
68
69from utils.oizom_logger import OizomLogger
70
71# -----------------------------------------------------------------------------
72# Configure logging
73# -----------------------------------------------------------------------------
74basic_logger = OizomLogger(__name__).get()
75context_logger = OizomLogger(__name__)
76
77# RGB CONFIGURATION
78pixel_pin = board.D18
79num_pixels = 10
80pixels = neopixel.NeoPixel(pixel_pin, num_pixels)
81beacon_config = {
82 "en": 1,
83 "sc": "aqi",
84 "range": [0, 50, 100, 200, 300, 400],
85 "colors": ["50f0e6", "29cc70", "f0e641", "d08000", "ff0000", "c700d0"],
86}
87beacon_flag = False
88
89# colors
90cyan = (0, 255, 255)
91red = (255, 0, 0)
92green = (0, 255, 0)
93blue = (0, 0, 255)
94magenta = (170, 10, 127)
95yellow = (255, 255, 0)
96white = (255, 255, 255)
97purple = (68, 0, 68)
98neonOrange = (255, 95, 31)
99unsupported_color = white
100
101# pattern
102pattern_fade = 0
103pattern_normal_blink = 1
104pattern_slow_blink = 2
105pattern_fast_blink = 3
106pattern_ultra_fast_blink = 4
107
108all_colors = [cyan, red, green, blue, magenta, yellow, white]
109
110# pattern { <network_status_indicator_number>: [<color>, <pattern>] }
111status = {
112 0: [green, pattern_normal_blink, "Disconnected"],
113 1: [cyan, pattern_fade, "Connected"],
114 2: [blue, pattern_slow_blink, "No Simcard"],
115 3: [magenta, pattern_fade, "Sending Data"],
116 4: [yellow, pattern_fast_blink, "Getting Config"],
117 5: [red, pattern_ultra_fast_blink, "Some Bug Found"],
118 6: [purple, pattern_ultra_fast_blink, "init success"],
119 7: [neonOrange, pattern_normal_blink, "OGS Allocation"],
120 8: [blue, pattern_fade, "QC Mode"],
121}
122
123
[docs]
124def hex_to_rgb(hex_code: str) -> tuple:
125 """Convert a 6-character hex color code to an ``(R, G, B)`` tuple.
126
127 Parses two characters at a time from ``hex_code`` (red, green, blue) and
128 returns 8-bit channel values suitable for assignment to a NeoPixel pixel.
129 Invalid lengths log an error via ``context_logger`` and fall back to
130 black so that the LED strip stays in a defined state.
131
132 Args:
133 hex_code (str): Six-character hex string (for example ``"50f0e6"``),
134 without a leading ``#``. Letters may be upper or lower case.
135
136 Returns:
137 tuple: ``(red, green, blue)`` integers, each in the inclusive range
138 ``0..255``. ``(0, 0, 0)`` is returned for malformed input.
139
140 Raises:
141 ValueError: If ``hex_code`` is the correct length but contains
142 characters that are not valid hexadecimal digits.
143
144 Example:
145 Convert the default "Connected" beacon color::
146
147 >>> from OzWrapper.OzRGB.OzRGB import hex_to_rgb # doctest: +SKIP
148 >>> hex_to_rgb("50f0e6") # doctest: +SKIP
149 (80, 240, 230)
150
151 Note:
152 The function does not accept a leading ``#`` or 3-digit shorthand
153 (``"fff"``). Callers must supply the canonical 6-character form
154 used by :data:`beacon_config`.
155 """
156
157 # Validate input
158 if len(hex_code) != 6:
159 context_logger.error_with_context("RGB", "Hex code must be 6 characters long.")
160 return (0, 0, 0)
161
162 # Convert each pair of hex digits to an integer
163 r = int(hex_code[0:2], 16)
164 g = int(hex_code[2:4], 16)
165 b = int(hex_code[4:6], 16)
166
167 return (r, g, b)
168
169
[docs]
170def get_color_from_value(value: int) -> str | None:
171 """Look up the beacon hex color for a given AQI value.
172
173 Walks the ordered ``beacon_config["range"]`` thresholds and returns the
174 color from ``beacon_config["colors"]`` whose lower bound is the largest
175 value ``<= value``. Values at or above the final threshold are clamped
176 to the last color, modelling the "hazardous" AQI bucket.
177
178 Args:
179 value (int): AQI (or other index) value to classify. Negative
180 values are treated as out-of-range and yield ``None``.
181
182 Returns:
183 str | None: Hex color string (without ``#`` prefix) from
184 ``beacon_config["colors"]``, or ``None`` if ``value`` is below the
185 first range boundary.
186
187 Raises:
188 KeyError: If ``beacon_config`` is missing the ``"range"`` or
189 ``"colors"`` keys.
190 IndexError: If ``range`` and ``colors`` are not the same length and
191 ``value`` falls in an undefined bucket.
192
193 Example:
194 Resolve a "moderate" AQI bucket::
195
196 >>> from OzWrapper.OzRGB.OzRGB import get_color_from_value # doctest: +SKIP
197 >>> get_color_from_value(75) # 50 <= 75 < 100 -> "f0e641" # doctest: +SKIP
198 'f0e641'
199
200 Note:
201 The lookup is half-open ``[lower, upper)`` for intermediate buckets
202 and closed ``[lower, +inf)`` for the final bucket. Pair this helper
203 with :func:`hex_to_rgb` before assigning to a NeoPixel pixel.
204 """
205 ranges = beacon_config["range"]
206 colors = beacon_config["colors"]
207
208 for i in range(len(ranges) - 1):
209 if ranges[i] <= value < ranges[i + 1]:
210 return colors[i]
211 if value >= ranges[-1]:
212 return colors[-1]
213 return None
214
215
216class Magic:
217 """LED animation engine providing fade and blink patterns for NeoPixel strips.
218
219 ``Magic`` owns a non-blocking color state machine over the attached
220 :class:`neopixel.NeoPixel` strip. Each call to :meth:`fade` or
221 :meth:`blinking` performs at most one brightness update, gated by the
222 millisecond clock returned by :meth:`millis`, so the caller's main loop
223 stays responsive while patterns are running.
224
225 Color state machine:
226 * **Fade**: ``counter_brightness`` sweeps between ``min_bright`` and
227 ``max_bright``, with ``_fade_direction`` flipping at each endpoint.
228 ``RISING`` segments wait ``fadInDelay`` ms between steps;
229 ``FALLING`` segments wait ``fadeOutWait`` ms.
230 * **Blink**: ``blink_direction`` toggles between full brightness
231 (``max_bright / bright_divider``) and ``0`` every ``blink_delay``
232 ms. Pattern speed is selected by the caller via
233 :data:`pattern_normal_blink`, :data:`pattern_slow_blink`,
234 :data:`pattern_fast_blink`, or :data:`pattern_ultra_fast_blink`.
235
236 Attributes:
237 brightness (float): Current brightness level (``0.0`` to ``1.0``)
238 mirrored onto ``pixels.brightness``.
239 min_bright (int): Minimum brightness counter value for fade
240 animations.
241 max_bright (int): Maximum brightness counter value for fade
242 animations.
243 bright_divider (int): Divisor that converts the integer counter to
244 the ``0.0-1.0`` NeoPixel brightness range.
245 fadInDelay (int): Milliseconds between brightness increments during
246 fade-in.
247 fadeOutWait (int): Milliseconds between brightness decrements
248 during fade-out.
249 pixels (neopixel.NeoPixel): NeoPixel strip instance being
250 controlled.
251 RISING (int): Direction constant (``1``) for increasing brightness.
252 FALLING (int): Direction constant (``-1``) for decreasing
253 brightness.
254 blink_delay (int): Milliseconds between blink toggles.
255 blink_direction (int): Current blink phase (``RISING`` or
256 ``FALLING``).
257 counter_brightness (int): Live counter driving fade output.
258 prev_blinking (int): Timestamp of the last blink toggle, in ms.
259
260 Example:
261 Drive a 100 ms blink on the first pixel::
262
263 >>> import neopixel, board # doctest: +SKIP
264 >>> from OzWrapper.OzRGB.OzRGB import Magic # doctest: +SKIP
265 >>> strip = neopixel.NeoPixel(board.D18, 10) # doctest: +SKIP
266 >>> m = Magic(strip) # doctest: +SKIP
267 >>> while True: # doctest: +SKIP
268 ... m.blinking(100)
269 """
270
271 brightness = 0
272 min_bright = 1
273 max_bright = 100
274 bright_divider = 100
275 fadInDelay = 20
276 fadeOutWait = 10
277 fadeOutWait = fadInDelay
278 pixels = None
279 _fade_current_time = None
280 _fade_prev_time = None
281 _fade_direction = 1
282
283 counter_brightness = 0
284
285 RISING = 1
286 FALLING = -1
287
288 blink_delay = 60
289 blink_direction = RISING
290 prev_blinking = 0
291
292 def __init__(self, pixels: neopixel.NeoPixel) -> None:
293 """Initialize the animation engine with a NeoPixel strip.
294
295 Stores the strip reference, primes fade parameters via
296 :meth:`setFadeProperty`, and seeds ``blink_delay`` to the default
297 ``60`` ms tick so the engine is immediately usable by :meth:`fade`
298 and :meth:`blinking`.
299
300 Args:
301 pixels (neopixel.NeoPixel): NeoPixel strip instance to control.
302 Must already be configured for the desired pin and length
303 (the module-level ``pixels`` object on ``board.D18`` is the
304 typical argument).
305
306 Returns:
307 None: This is a constructor and has no return value.
308
309 Raises:
310 AttributeError: If ``pixels`` does not expose the
311 :attr:`brightness` and :meth:`show` interface expected by
312 :meth:`fade` and :meth:`blinking`.
313
314 Example:
315 Create an engine on the module-level NeoPixel strip::
316
317 >>> from OzWrapper.OzRGB.OzRGB import Magic, pixels # doctest: +SKIP
318 >>> engine = Magic(pixels) # doctest: +SKIP
319
320 Note:
321 Construction does **not** clear or fill the strip. The first
322 visual change occurs only when :meth:`fade` or :meth:`blinking`
323 is called from the main loop.
324 """
325 self.pixels = pixels
326 self.setFadeProperty()
327 self.blink_delay = 60
328
329 def setFadeProperty(
330 self,
331 brightness: int = 0,
332 min_bright: int = 1,
333 max_bright: int = 100,
334 bright_divider: int = 100,
335 fadInDelay: int = 20,
336 fadeOutWait: int = 10,
337 ) -> None:
338 """Configure fade animation properties.
339
340 Resets the fade state machine: timestamps are re-seeded from
341 :meth:`millis`, the direction is forced to ``RISING``, and the
342 strip brightness is pulled down to ``min_bright / bright_divider``
343 so subsequent :meth:`fade` ticks begin from a known floor.
344
345 Args:
346 brightness (int): Initial brightness counter value used for the
347 ``brightness`` attribute before the floor override.
348 min_bright (int): Minimum brightness counter for the fade range.
349 Used as the dim endpoint and as the divisor for the initial
350 strip brightness.
351 max_bright (int): Maximum brightness counter for the fade
352 range. Used as the bright endpoint.
353 bright_divider (int): Divisor that converts the integer counter
354 to the NeoPixel ``0.0-1.0`` brightness range.
355 fadInDelay (int): Milliseconds between fade-in steps.
356 fadeOutWait (int): Milliseconds between fade-out steps.
357
358 Returns:
359 None: Mutates ``self`` in place; no value is returned.
360
361 Raises:
362 ZeroDivisionError: If ``bright_divider`` is ``0``.
363 AttributeError: If ``self.pixels`` has not been assigned (the
364 constructor normally guarantees this).
365
366 Example:
367 Slow the fade-in to ~50 ms per step::
368
369 >>> engine.setFadeProperty(fadInDelay=50) # doctest: +SKIP
370
371 Note:
372 Calling this mid-animation snaps brightness back to the floor
373 and may produce a visible flicker. Prefer to set properties
374 once during initialization.
375 """
376 self.brightness = brightness
377 self.min_bright = min_bright
378 self.max_bright = max_bright
379 self.bright_divider = bright_divider
380 self.fadInDelay = fadInDelay
381 self.fadeOutWait = fadeOutWait
382 self._fade_current_time = self.millis()
383 self._fade_prev_time = self.millis()
384 self._fade_direction = 1
385 self.brightness = self.min_bright / self.bright_divider # OVERRIDE
386 self.pixels.brightness = self.brightness
387 self.counter_brightness = self.min_bright + 1
388
389 def fade(self) -> None:
390 """Advance the fade animation by one step (non-blocking).
391
392 Flips ``_fade_direction`` whenever ``counter_brightness`` hits
393 ``min_bright`` or ``max_bright``, then — if the appropriate
394 per-direction delay (``fadInDelay`` or ``fadeOutWait``) has elapsed
395 since the last step — increments or decrements the counter,
396 writes the scaled value to ``pixels.brightness``, and calls
397 ``pixels.show()``.
398
399 Args:
400 None: This method takes no arguments; it operates on the
401 instance state mutated by :meth:`setFadeProperty`.
402
403 Returns:
404 None: The strip is updated as a side effect; nothing is
405 returned.
406
407 Raises:
408 AttributeError: If the engine has not been initialized via
409 :meth:`setFadeProperty` (timestamps would be ``None``).
410
411 Example:
412 Drive a continuous breathing pattern from the main loop::
413
414 >>> while True: # doctest: +SKIP
415 ... engine.fade()
416
417 Note:
418 This call returns immediately when the per-direction delay has
419 not yet elapsed, so it is safe to invoke at high frequency.
420 Combine with :meth:`blinking` only by switching the active
421 pattern via the module-level ``status`` table — never call
422 both in the same tick.
423 """
424 self._fade_current_time = self.millis()
425 if self._fade_direction == self.RISING and self.counter_brightness >= self.max_bright:
426 self._fade_direction = self.FALLING
427 elif self._fade_direction == self.FALLING and self.counter_brightness <= self.min_bright:
428 self._fade_direction = self.RISING
429
430 __dif_time = self._fade_current_time - self._fade_prev_time
431
432 if (self._fade_direction == self.RISING and (__dif_time) >= self.fadInDelay) or (
433 self._fade_direction == self.FALLING and (__dif_time) >= self.fadeOutWait
434 ):
435 _b_print = self.counter_brightness / self.bright_divider
436 self.counter_brightness += self._fade_direction
437 self.pixels.brightness = self.counter_brightness / self.bright_divider
438 self.pixels.show()
439 self._fade_prev_time = self._fade_current_time
440
441 def blinking(self, blink_delay: int = 60) -> None:
442 """Advance the blink animation by one step (non-blocking).
443
444 If at least ``blink_delay`` ms have passed since the previous
445 toggle, swaps ``blink_direction`` and sets ``pixels.brightness`` to
446 either ``max_bright / bright_divider`` (when transitioning from
447 ``RISING``) or ``0`` (when transitioning from ``FALLING``).
448 ``pixels.show()`` is always called so a freshly updated pixel
449 color from :func:`main` becomes visible immediately.
450
451 Args:
452 blink_delay (int): Milliseconds between on/off toggles. The
453 caller selects ``100`` for normal, ``600`` for slow, ``70``
454 for fast, and ``30`` for ultra-fast blink patterns.
455
456 Returns:
457 None: Updates the strip in place; no value is returned.
458
459 Raises:
460 AttributeError: If ``self.pixels`` is ``None`` (engine was not
461 constructed via :meth:`__init__`).
462
463 Example:
464 Trigger a 30 ms ultra-fast blink for a critical alarm::
465
466 >>> engine.blinking(30) # doctest: +SKIP
467
468 Note:
469 Unlike :meth:`fade`, this method always issues ``pixels.show()``
470 even when no brightness toggle occurred, which is what keeps
471 the AQI beacon ring (pixels ``1..N-1``) refreshed on every main
472 loop iteration.
473 """
474 self.blink_delay = blink_delay
475 __current_blink = self.millis()
476 __dif_time = __current_blink - self.prev_blinking
477 if (__dif_time) > self.blink_delay:
478 if self.blink_direction == self.RISING:
479 self.pixels.brightness = self.max_bright / self.bright_divider
480 self.blink_direction = self.FALLING
481 self.prev_blinking = __current_blink
482 elif self.blink_direction == self.FALLING:
483 self.pixels.brightness = 0
484 self.blink_direction = self.RISING
485 self.prev_blinking = __current_blink
486 self.pixels.show()
487
488 def micros(self) -> int:
489 """Return the current time in microseconds.
490
491 Thin Arduino-style wrapper around :func:`time.time` that returns an
492 integer number of microseconds since the Unix epoch. Provided as a
493 sibling to :meth:`millis` for callers that need finer-grained
494 scheduling than the millisecond fade/blink loops use.
495
496 Args:
497 None: This method takes no arguments.
498
499 Returns:
500 int: Microseconds since the Unix epoch.
501
502 Raises:
503 OSError: Propagated from :func:`time.time` if the underlying
504 system clock cannot be read.
505
506 Example:
507 Measure the cost of a single fade step::
508
509 >>> t0 = engine.micros() # doctest: +SKIP
510 >>> engine.fade() # doctest: +SKIP
511 >>> elapsed_us = engine.micros() - t0 # doctest: +SKIP
512
513 Note:
514 Resolution is bounded by the host clock; on Raspberry Pi this
515 is sub-microsecond but not monotonic — do **not** use this for
516 interval timing that must survive an NTP step. Use
517 :func:`time.monotonic` if a monotonic source is required.
518 """
519 return int(time.time() * 1000000)
520
521 def millis(self) -> int:
522 """Return the current time in milliseconds.
523
524 Arduino-style wall-clock helper used internally by :meth:`fade`
525 and :meth:`blinking` to gate non-blocking animation steps.
526
527 Args:
528 None: This method takes no arguments.
529
530 Returns:
531 int: Milliseconds since the Unix epoch.
532
533 Raises:
534 OSError: Propagated from :func:`time.time` if the underlying
535 system clock cannot be read.
536
537 Example:
538 Stamp the start of an animation cycle::
539
540 >>> start_ms = engine.millis() # doctest: +SKIP
541
542 Note:
543 Like :meth:`micros`, this is **not** monotonic. The fade/blink
544 state machines tolerate the occasional NTP correction because
545 each tick only checks a small delta; do not rely on it for
546 long-duration interval measurement.
547 """
548 return int(time.time() * 1000)
549
550
[docs]
551def main(
552 os_network_alert_file: None,
553 network_status: Queue | None = None,
554 beacon_queue: Queue | None = None,
555) -> None:
556 """Run the RGB LED main loop, updating status and beacon colors continuously.
557
558 Instantiates a :class:`Magic` engine over the module-level
559 :data:`pixels` strip, then enters an infinite loop that:
560
561 * peeks the head of ``network_status`` to resolve the current status
562 code, looks up ``(color, pattern, label)`` in :data:`status`, and
563 writes the color to pixel ``0``;
564 * when :data:`beacon_flag` is ``True``, drains ``beacon_queue`` for an
565 AQI payload, resolves a beacon color via :func:`get_color_from_value`
566 and :func:`hex_to_rgb`, and fills pixels ``1..9`` with it;
567 * dispatches to :meth:`Magic.blinking` or :meth:`Magic.fade` to animate
568 pixel ``0`` according to the selected pattern.
569
570 Any exception inside the loop fills the strip with
571 :data:`unsupported_color` (white), runs a fade tick to keep motion
572 visible, and rate-limits the error log to one line per
573 ``print_interval`` second.
574
575 Args:
576 os_network_alert_file (None): Unused. Retained for backward
577 compatibility with the legacy entry-point signature that took a
578 network alert file path.
579 network_status (queue.Queue | None): Queue providing integer
580 network status codes for LED color and pattern. Codes are
581 looked up in :data:`status`; values from ``0`` (Disconnected)
582 through ``8`` (QC Mode) are recognised.
583 beacon_queue (queue.Queue | None): Queue providing AQI data for
584 beacon LED coloring. Each element is a dict that either
585 contains the ``"aqi"`` key directly or nests sensor channels
586 under ``"d"``.
587
588 Returns:
589 None: This function never returns under normal operation; the
590 outer ``while 1`` loop runs for the lifetime of the process.
591
592 Raises:
593 AttributeError: If ``network_status`` is ``None`` and the queue
594 peek is attempted (the wider ``except`` catches and logs this).
595 KeyboardInterrupt: Propagated when the operator stops the process.
596
597 Example:
598 Spawn the indicator from a parent supervisor::
599
600 >>> from multiprocessing import Process # doctest: +SKIP
601 >>> from queue import Queue # doctest: +SKIP
602 >>> from OzWrapper.OzRGB.OzRGB import main # doctest: +SKIP
603 >>> net_q, beacon_q = Queue(), Queue() # doctest: +SKIP
604 >>> p = Process(target=main, args=(None, net_q, beacon_q)) # doctest: +SKIP
605 >>> p.start() # doctest: +SKIP
606
607 Note:
608 The loop is intentionally tight (no ``time.sleep``) because
609 :meth:`Magic.fade` and :meth:`Magic.blinking` self-throttle on
610 :meth:`Magic.millis`. Driving the WS281x line at 800 kHz from
611 Python requires the loop to keep calling ``pixels.show()`` so the
612 kernel-side DMA stays fed; never insert ``sleep`` here.
613
614 See Also:
615 :class:`SensorBase.SensorBase.GenericSensor`: Sibling sensor
616 wrappers consume the same status queue idiom used here.
617 """
618 current_time = int(time.time())
619 prev_time = int(time.time())
620 print_interval = 1
621 f = Magic(pixels=pixels)
622 beacon_color = unsupported_color
623 value = -1
624 while 1:
625 output = -1
626 current_time = int(time.time())
627 try:
628 output = network_status.queue[0]
629 # output = network_status[0]
630 status.get(output)[0]
631 except Exception:
632 try:
633 output = network_status.queue[1]
634 except Exception as e:
635 context_logger.error_with_context("RGB", f"main: {e}")
636 try:
637 # Get all the configuration
638 _color = status.get(output)[0] # color
639 _pattern = status.get(output)[1]
640
641 # print network status output
642 if (current_time - prev_time) > print_interval:
643 # print("Network : {0}".format(_output_string))
644 prev_time = int(time.time())
645
646 pixels[0] = _color
647
648 if beacon_flag:
649 try:
650 if not beacon_queue.empty():
651 data = beacon_queue.queue[0]
652 parameter = beacon_config["sc"]
653 if parameter == "aqi":
654 if data and parameter in data:
655 value = data[parameter]
656 else:
657 if data and parameter in data["d"]:
658 value = data["d"][parameter]
659 if value >= 0:
660 beacon_color = hex_to_rgb(get_color_from_value(value))
661 context_logger.info_with_context(
662 "BEACON",
663 f"value: {value} color: {get_color_from_value(value)}",
664 )
665 else:
666 beacon_color = unsupported_color
667 beacon_queue.queue.clear()
668
669 # Update Beacon color
670 pixels[1:] = [beacon_color] * 9
671 except Exception as e:
672 context_logger.error_with_context("BEACON", f"update: {e}")
673
674 # run patterns
675 if _pattern == pattern_normal_blink:
676 f.blinking(100)
677 elif _pattern == pattern_slow_blink:
678 f.blinking(600)
679 elif _pattern == pattern_fast_blink:
680 f.blinking(70)
681 elif _pattern == pattern_ultra_fast_blink:
682 f.blinking(30)
683 elif _pattern == pattern_fade:
684 f.fade()
685
686 except Exception as e:
687 pixels.fill(unsupported_color)
688 f.fade()
689 if (current_time - prev_time) > print_interval:
690 context_logger.error_with_context("RGB", f"main: {e}")
691 prev_time = int(time.time())
692
693
694if __name__ == "__main__":
695 NETWORK_FILE = "/etc/ozone_network.txt"
696 main(NETWORK_FILE)