OzWrapper.OzRGB.OzRGB¶

RGB LED indicator wrapper for NeoPixel status display.

Drives a NeoPixel LED strip to indicate device status (network connectivity, data sending, errors) through color and blink/fade patterns, and optionally shows AQI-based beacon colors on remaining LEDs. Pixel 0 carries the status color/pattern produced by the Magic animation engine; pixels 1..N-1 carry the beacon color resolved from the live AQI queue.

The module also exposes lightweight helpers hex_to_rgb() and get_color_from_value() for converting hex strings and mapping AQI values to beacon colors, plus a main() event loop that the parent process spawns as a long-running indicator worker.

Example

Run the indicator loop with a network status queue and beacon queue:

>>> from queue import Queue
>>> from OzWrapper.OzRGB import OzRGB
>>> net_q, beacon_q = Queue(), Queue()
>>> net_q.put(1)  # 1 -> Connected (cyan, fade)
>>> OzRGB.main(None, net_q, beacon_q)

Note

The NeoPixel data line is driven via the on-board PWM peripheral on GPIO18 (board.D18); the NeoPixel WS281x protocol requires an 800 kHz signal with sub-microsecond timing, so the pin and frequency must not be reconfigured at runtime. Ten LEDs are addressed by default (num_pixels = 10); index 0 is reserved for status and indices 1..9 form the AQI beacon ring.

Reference config:

{
  "beacon": {
    "en": 1,
    "sc": "aqi",
    "range": [
      0,
      50,
      100,
      200,
      300,
      400
    ],
    "colors": [
      "50f0e6",
      "29cc70",
      "f0e641",
      "d08000",
      "ff0000",
      "c700d0"
    ]
  }
}

Classes¶

Magic

LED animation engine providing fade and blink patterns for NeoPixel strips.

Functions¶

`get_color_from_value`(→ str \| None)	Look up the beacon hex color for a given AQI value.
`hex_to_rgb`(→ tuple)	Convert a 6-character hex color code to an `(R, G, B)` tuple.
`main`(→ None)	Run the RGB LED main loop, updating status and beacon colors continuously.

Module Contents¶

OzWrapper.OzRGB.OzRGB.get_color_from_value(value: int) → str | None[source]¶

Look up the beacon hex color for a given AQI value.

Walks the ordered beacon_config["range"] thresholds and returns the color from beacon_config["colors"] whose lower bound is the largest value <= value. Values at or above the final threshold are clamped to the last color, modelling the “hazardous” AQI bucket.

Parameters:

value (int) – AQI (or other index) value to classify. Negative values are treated as out-of-range and yield None.

Returns:

Hex color string (without # prefix) from beacon_config["colors"], or None if value is below the first range boundary.

Return type:

str | None

Raises:

KeyError – If beacon_config is missing the "range" or "colors" keys.
IndexError – If range and colors are not the same length and value falls in an undefined bucket.

Example

Resolve a “moderate” AQI bucket:

>>> from OzWrapper.OzRGB.OzRGB import get_color_from_value
>>> get_color_from_value(75)  # 50 <= 75 < 100 -> "f0e641"
'f0e641'

Note

The lookup is half-open [lower, upper) for intermediate buckets and closed [lower, +inf) for the final bucket. Pair this helper with hex_to_rgb() before assigning to a NeoPixel pixel.

OzWrapper.OzRGB.OzRGB.hex_to_rgb(hex_code: str) → tuple[source]¶

Convert a 6-character hex color code to an (R, G, B) tuple.

Parses two characters at a time from hex_code (red, green, blue) and returns 8-bit channel values suitable for assignment to a NeoPixel pixel. Invalid lengths log an error via context_logger and fall back to black so that the LED strip stays in a defined state.

Parameters:: hex_code (str) – Six-character hex string (for example "50f0e6"), without a leading #. Letters may be upper or lower case.
Returns:: (red, green, blue) integers, each in the inclusive range 0..255. (0, 0, 0) is returned for malformed input.
Return type:: tuple
Raises:: ValueError – If hex_code is the correct length but contains characters that are not valid hexadecimal digits.

Example

Convert the default “Connected” beacon color:

>>> from OzWrapper.OzRGB.OzRGB import hex_to_rgb
>>> hex_to_rgb("50f0e6")
(80, 240, 230)

Note

The function does not accept a leading # or 3-digit shorthand ("fff"). Callers must supply the canonical 6-character form used by beacon_config.

OzWrapper.OzRGB.OzRGB.main(os_network_alert_file: None, network_status: queue.Queue | None = None, beacon_queue: queue.Queue | None = None) → None[source]¶

Run the RGB LED main loop, updating status and beacon colors continuously.

Instantiates a Magic engine over the module-level pixels strip, then enters an infinite loop that:

peeks the head of network_status to resolve the current status code, looks up (color, pattern, label) in status, and writes the color to pixel 0;
when beacon_flag is True, drains beacon_queue for an AQI payload, resolves a beacon color via get_color_from_value() and hex_to_rgb(), and fills pixels 1..9 with it;
dispatches to Magic.blinking() or Magic.fade() to animate pixel 0 according to the selected pattern.

Any exception inside the loop fills the strip with unsupported_color (white), runs a fade tick to keep motion visible, and rate-limits the error log to one line per print_interval second.

Parameters:

os_network_alert_file (None) – Unused. Retained for backward compatibility with the legacy entry-point signature that took a network alert file path.
network_status (queue.Queue | None) – Queue providing integer network status codes for LED color and pattern. Codes are looked up in status; values from 0 (Disconnected) through 8 (QC Mode) are recognised.
beacon_queue (queue.Queue | None) – Queue providing AQI data for beacon LED coloring. Each element is a dict that either contains the "aqi" key directly or nests sensor channels under "d".

Returns:

This function never returns under normal operation; the outer while 1 loop runs for the lifetime of the process.

Return type:

None

Raises:

AttributeError – If network_status is None and the queue peek is attempted (the wider except catches and logs this).
KeyboardInterrupt – Propagated when the operator stops the process.

Example

Spawn the indicator from a parent supervisor:

>>> from multiprocessing import Process
>>> from queue import Queue
>>> from OzWrapper.OzRGB.OzRGB import main
>>> net_q, beacon_q = Queue(), Queue()
>>> p = Process(target=main, args=(None, net_q, beacon_q))
>>> p.start()

Note

The loop is intentionally tight (no time.sleep) because Magic.fade() and Magic.blinking() self-throttle on Magic.millis(). Driving the WS281x line at 800 kHz from Python requires the loop to keep calling pixels.show() so the kernel-side DMA stays fed; never insert sleep here.

`NETWORK_FILE`
`all_colors`
`basic_logger`
`beacon_config`
`beacon_flag`
`blue`
`context_logger`
`cyan`
`green`
`magenta`
`neonOrange`
`num_pixels`
`pattern_fade`
`pattern_fast_blink`
`pattern_normal_blink`
`pattern_slow_blink`
`pattern_ultra_fast_blink`
`pixel_pin`
`pixels`
`purple`
`red`
`status`
`unsupported_color`
`white`
`yellow`

OzWrapper.OzRGB.OzRGB¶

Attributes¶

Classes¶

Functions¶

Module Contents¶