"""Wire protocol for the esp324xlight firmware.
The firmware exposes a line-oriented command surface on TCP 1337. The only
commands this addon emits are `pwm`, `pwm2`, `stopfade`, `gamma` and `freq`.
Channel order is B, G, R -- not RGB. This is not a quirk of the Blender side:
the firmware caps every value at index `count % 3 == 2` to RED_MAX_PWM, so the
third slot of each triple *is* the red channel, and `gamma` prints its three
arguments back as "B=.. G=.. R=..". The original quadpanel.py emitted
color[2], color[1], color[0] for the same reason.
A fixture is 12 channels = 4 slots x 3 channels. Slots with no light assigned
are padded with `0 0 0`; with INVERT_HIGH_AND_LOW in the firmware, input 0 is
LED-off, so padding is dark rather than full-on.
"""
from collections import namedtuple
CHANNELS_PER_SLOT = 3
SLOTS_PER_FIXTURE = 4
NUM_CHANNELS = CHANNELS_PER_SLOT * SLOTS_PER_FIXTURE
# Firmware absolute ceilings (esp324xlight.ino: MAX_PWM / RED_MAX_PWM).
# parsePWM() rejects the WHOLE command if any single value exceeds these, so we
# always clamp before sending -- one bad value would drop all 12 channels.
FIRMWARE_MAX_PWM = 65535
FIRMWARE_RED_MAX_PWM = 45000
def clamp(v, lo, hi):
return lo if v < lo else (hi if v > hi else v)
def pwm_from_unit(value, pwm_max):
"""Map a 0..1 Blender channel to a 0..pwm_max integer PWM input."""
return int(clamp(round(abs(value) * pwm_max), 0, pwm_max))
def channels_from_rgbe(r, g, b, energy, red_max, use_energy, energy_ref):
"""[B, G, R] PWM ints from raw channel values.
Split out from slot_channels so the same conversion serves both the current
frame (read off the datablock) and a future keyframe (evaluated off FCurves).
"""
if use_energy:
# Scale colour by normalised energy so animators can key intensity the
# way they normally would. energy_ref is what counts as "full".
ref = energy_ref if energy_ref > 0.0 else 1.0
scale = clamp(energy / ref, 0.0, 1.0)
r, g, b = r * scale, g * scale, b * scale
return [
pwm_from_unit(b, FIRMWARE_MAX_PWM),
pwm_from_unit(g, FIRMWARE_MAX_PWM),
pwm_from_unit(r, min(red_max, FIRMWARE_RED_MAX_PWM)),
]
def slot_channels(light_obj, red_max, use_energy, energy_ref):
"""Return [B, G, R] PWM ints for one light object, or None if unusable."""
data = getattr(light_obj, "data", None)
if data is None or not hasattr(data, "color"):
return None
return channels_from_rgbe(
data.color[0], data.color[1], data.color[2],
getattr(data, "energy", energy_ref),
red_max, use_energy, energy_ref)
def frame_values(slot_map, red_max, use_energy, energy_ref):
"""Build the 12 channel values for a fixture from {slot: light_object}."""
values = []
for slot in range(1, SLOTS_PER_FIXTURE + 1):
obj = slot_map.get(slot)
chans = slot_channels(obj, red_max, use_energy, energy_ref) if obj else None
values.extend(chans if chans else [0, 0, 0])
return values
def pwm_command(values):
return "pwm " + " ".join(str(v) for v in values) + "\n"
def pick_pwm2_args(total_ms):
"""Split a fade duration into (numPeriods, periodMs), both within 1..1000.
updateFade() interpolates on wall-clock -- t = elapsed / totalMs, recomputed
every loop() -- so numPeriods does NOT quantise the fade. The pair only has
to multiply out to the intended duration while both stay inside the range
parsePWM enforces. Total is therefore capped at 1000*1000 ms.
"""
total_ms = int(clamp(total_ms, 1, 1000 * 1000))
if total_ms <= 1000:
return 1, total_ms
period_ms = (total_ms + 999) // 1000 # ceil, so numPeriods stays <= 1000
num_periods = max(1, total_ms // period_ms)
return num_periods, period_ms
def pwm2_command(values, fade_ms):
"""Interpolated move.
The firmware fades in perceptual space and keeps PinCurrentValues live, so a
new pwm2 mid-fade picks up from where the LEDs actually are.
"""
periods, period_ms = pick_pwm2_args(fade_ms)
return ("pwm2 %d %d " % (periods, period_ms)) + " ".join(str(v) for v in values) + "\n"
def gamma_command(b, g, r):
# Firmware silently ignores any value outside 0.5..5.0.
return "gamma %.2f %.2f %.2f\n" % (
clamp(b, 0.5, 5.0), clamp(g, 0.5, 5.0), clamp(r, 0.5, 5.0))
def freq_command(hz):
return "freq %d\n" % int(clamp(hz, 1, 40000))
def blackout_command():
return "stopfade\n" + pwm_command([0] * NUM_CHANNELS)
# ---------------------------------------------------------------------------
# What the engine hands the sender: the fixture's 12 values and how long to
# fade to them (0 = instantly). The sender turns it into whatever the unit
# speaks -- a whole-unit pwm/pwm2 for proto <= 2, per-slot grants and `slot`
# lines for proto 3 (docs/DESIGN-collab.md §5). Comparable, so identical
# consecutive frames are skipped.
Frame = namedtuple("Frame", "values fade_ms")
def frame_command(frame):
"""Proto <= 2: one whole-unit command."""
values = list(frame.values)
if frame.fade_ms > 0:
return pwm2_command(values, frame.fade_ms)
return pwm_command(values)
# ---- proto 3: shared control ----------------------------------------------
def grant_command(slot, holder):
return "grant %s %s\n" % (slot, holder)
def slot_command(slot, epoch, n, bgr, fade_ms):
return "slot %d %d %d %d %d %d %d\n" % (slot, epoch, n, bgr[0], bgr[1], bgr[2], int(fade_ms))
def release_command(slot):
return "release %s\n" % slot