Discovery not working - defining using MAC & IP

[HaberHash]

Given that discovering bulbs is so problematic can someone share how we'd define a list of lights using MAC & IP? I've tried several combinations based on what's in the readme, but can't seem to get it working properly.

Many Thanks!!!

[mclarkk]

From the README:

lights = lan.get_lights()                              # Option 1: Discovery
light = Light("12:34:56:78:9a:bc", "192.168.1.42")     # Option 2: Direct

If that doesn't help, please post your actual code and the errors you're getting.

[HaberHash]

Here is what I've implemented, but it's still not using defined lights:

import time
import colorsys
import signal
import sys
import sacn
import asyncio
from lifxlan import LifxLAN
from lifxlan.light import Light
#import threading

def create_manual_light_list(ip_addresses=None, mac_addresses=None, labels=None):
    """Creates a list of Lifx Light objects manually."""
    if not ip_addresses and not mac_addresses and not labels:
        return []

    lists = [ip_addresses, mac_addresses, labels]
    lengths = [len(lst) if lst else 0 for lst in lists]
    max_len = max(lengths)

    if any(length > 0 and length != max_len for length in lengths):
        print("Warning: Input lists have different lengths. Using the longest list and padding with None.")

    lights = []
    for i in range(max_len):
        ip = ip_addresses[i] if ip_addresses and i < len(ip_addresses) else None
        mac = mac_addresses[i] if mac_addresses and i < len(mac_addresses) else None
        label = labels[i] if labels and i < len(labels) else None

        if ip or mac:
          lights.append(Light(ip, mac, label))
        else:
          print(f"Warning: Light at index {i} could not be created as neither IP nor MAC address were provided.")

    return lights

# List of LIFX Devices
manual_ip_list = ["10.10.1.1", "10.10.1.2", "10.10.1.3", "10.10.1.4"]
manual_mac_list =  ["d0:73:d5:11:11:11", "d0:73:d5:11:11:22", "d0:73:d5:11:11:33", "d0:73:d5:11:11:44"]
manual_label_list = ["Light1", "Light2", "Light3", "Light4"]

manual_lights = create_manual_light_list(manual_ip_list, manual_mac_list, manual_label_list)

async def set_light_color_async(light, color, duration=0.0):
    """Sets the color of a LIFX light asynchronously."""
    try:
        await asyncio.to_thread(light.set_color, color, duration=duration, rapid=True) # rapid=True for faster updates
        #print(f"Color set to {color} on {light.label} asynchronously.") #Optional print statement
    except Exception as e:
        print(f"Error setting color for {light.label}: {e}")


async def process_dmx_data(packet, lights):
    """Processes DMX data and sets light colors asynchronously."""
    tasks = []
    for idx, light in enumerate(lights):
        start, end = dmx_channels(idx)
        rgb = packet.dmxData[start:end]
        hsb = [int(x * scale_up) for x in colorsys.rgb_to_hsv(*[c / max_rgb_per_color for c in rgb])]

        if sum(rgb) > override_max_bright_total_rgb:
            adjust_perc = (sum(rgb) - override_max_bright_total_rgb) * override_scaling
            bright_adj = max_bright + ((max_bright_override - max_bright) * adjust_perc)
        else:
            bright_adj = max_bright

        hsb[2] = round(hsb[2] * bright_adj)  # Adjust the brightness
        color = tuple(hsb) + (3500,)  # Add kelvin
        #print(f"{light.label:20} set to {color} from RGB{rgb} - Adjust % {adjust_perc} bright_adj {bright_adj}") #Optional print statement

        tasks.append(set_light_color_async(light, color)) #Create a list of tasks

    await asyncio.gather(*tasks)  # Run all color setting tasks concurrently

def main():
    lan = LifxLAN()  # Initialize LifxLAN
    lights = manual_lights

    if not lights:
        print("No lights found (manual or discovered). Exiting.")
        sys.exit(1)

    # ... (rest of the configuration and setup remains the same)

    receiver = sacn.sACNreceiver()
    receiver.start()

    @receiver.listen_on('universe', universe=UNIVERSE)
    def callback(packet):
        asyncio.run(process_dmx_data(packet, lights)) # Run the async function
            
########## START CONFIG SETTINGS ###########

UNIVERSE = 3 # Which dmx universe it should be listening on

first_channel = 0       # First DMX channel - 1 (to account for array position)

numb_lights = 50     # How many lights you have on your network?  All must be in the DMX data

max_bright = 0.8 # Brightness, from 0 - 1

# If r + g + b > this number, adjust max_bright by the percentage between the override and 3 * max_rgb_per_color
# i.e. for 255 * 3, max_bright will be set to max_bright_override
# for 200 * 3 or lower, max_bright will not be adjusted
# for in between, will be adjusted proportionally
override_max_bright_total_rgb = 200 * 3
max_bright_override = 1.0

########### END CONFIG SETTINGS ###########

channels_per_light = 3  # Red, Green, Blue
scale_up = 65535 # This value will change what's concidered max brightness. minimum is 1 maximum is 65535
max_rgb_per_color = 255
max_rgb_total = max_rgb_per_color * 3
override_scaling = 1/(max_rgb_total - override_max_bright_total_rgb)
# test power control
print("Discovering lights...")
lifx = LifxLAN(numb_lights)
lights = lifx.get_lights()
original_powers = lifx.get_power_all_lights()
original_colors = lifx.get_color_all_lights()

def dmx_channels(idx):
    start = first_channel + idx * channels_per_light
    end = start + channels_per_light
    return start, end

lights.sort(key=lambda l: l.get_label())
for i, l in enumerate(lights):
    startChannel, endChannel = dmx_channels(i)
    print(f"{l.label} is assigned to DMX channels {startChannel + 1} to {endChannel}")

print("Running e1.31 Lifx reciever")
# provide an IP-Address to bind to if you are using Windows and want to use multicast
if len(sys.argv) > 1:
    receiver = sacn.sACNreceiver(sys.argv[1])
else:
    receiver = sacn.sACNreceiver()
receiver.start()  # start the receiving thread

# define a callback function
@receiver.listen_on('universe', universe=UNIVERSE)  # listens on universe 1
def callback(packet):  # packet type: sacn.DataPacket
    print("Recieved DMX data: ==> ",packet.dmxData[(first_channel):(first_channel + numb_lights*channels_per_light)])
    for idx, light in enumerate(lights):
        # color is [Hue, Saturation, Brightness, Kelvin], duration in ms
        start, end = dmx_channels(idx)
        rgb = packet.dmxData[start:end]
        hsb = [int(x * scale_up) for x in colorsys.rgb_to_hsv(*[c / max_rgb_per_color for c in rgb])]
        if sum(rgb) > override_max_bright_total_rgb:
            adjust_perc =  (sum(rgb) - override_max_bright_total_rgb) * override_scaling
            bright_adj = max_bright + ((max_bright_override - max_bright) * adjust_perc)
        else:
            adjust_perc = 0
            bright_adj = max_bright
        hsb[2] = round(hsb[2] * bright_adj) # Adjust the brightness
        color = tuple(hsb) + (3500,) # Add kelvin
        print(f"{light.label:20} set to {color} from RGB{rgb} - Adjust % {adjust_perc} bright_adj {bright_adj}")
        light.set_color(color, 0, True)

# optional: if you want to use multicast use this function with the universe as parameter
receiver.join_multicast(UNIVERSE)

def cleanup():
    receiver.stop()
    print("Restoring original colors and power to all lights...")
    for light in lights:
        light.set_color(original_colors[light])
        light.set_power(original_powers[light])

def signal_handler(sig, frame):
    print('You pressed Ctrl+C!')
    sys.exit(0)  # Exit gracefully

signal.signal(signal.SIGINT, signal_handler)
print("Press Ctrl+C to quit")

if __name__ == '__main__':
    main()

[mclarkk]

Can you post the simplest example that fails, the output, and describe what products you're communicating with and your network setup?

[HaberHash]

I have 47 devices. Most of the time it takes many tries to get all 47 to discover, thus the reason why I've been trying to figure out how I can easily hard-code a list of devices.

When I run the code below it's rare that all 47 get returned. I have 47 LIFX recessed lights. The types are in products.py and they do all eventually auto discover. My network is UniFi. Devices are all static IP and on an isolated lighting network. There are no network issues. I can certainly reconfigure if required, but my setup does work and is rock solid. Discovery has always been super problematic.

Even your verbose_lan.py doesn't discover all devices every time. It doesn't even discover 100% of the time.

import time
import colorsys
#import threading
import signal
import sys

import sacn
from lifxlan import LifxLAN


########## START CONFIG SETTINGS ###########

UNIVERSE = 3 # Which dmx universe it should be listening on

first_channel = 0       # First DMX channel - 1 (to account for array position)

numb_lights = 47     # How many lights you have on your network?  All must be in the DMX data

max_bright = 0.8 # Brightness, from 0 - 1

# If r + g + b > this number, adjust max_bright by the percentage between the override and 3 * max_rgb_per_color
# i.e. for 255 * 3, max_bright will be set to max_bright_override
# for 200 * 3 or lower, max_bright will not be adjusted
# for in between, will be adjusted proportionally
override_max_bright_total_rgb = 200 * 3
max_bright_override = 1.0

########### END CONFIG SETTINGS ###########

channels_per_light = 3  # Red, Green, Blue
scale_up = 65535 # This value will change what's concidered max brightness. minimum is 1 maximum is 65535
max_rgb_per_color = 255
max_rgb_total = max_rgb_per_color * 3
override_scaling = 1/(max_rgb_total - override_max_bright_total_rgb)
# test power control
print("Discovering lights...")
lifx = LifxLAN(numb_lights)
lights = lifx.get_lights()
# original_powers = lifx.get_power_all_lights()
# original_colors = lifx.get_color_all_lights()

def dmx_channels(idx):
    start = first_channel + idx * channels_per_light
    end = start + channels_per_light
    return start, end

lights.sort(key=lambda l: l.get_label())
for i, l in enumerate(lights):
    startChannel, endChannel = dmx_channels(i)
    print(f"{l.label} is assigned to DMX channels {startChannel + 1} to {endChannel}")

print("Running e1.31 Lifx reciever")
# provide an IP-Address to bind to if you are using Windows and want to use multicast
if len(sys.argv) > 1:
    receiver = sacn.sACNreceiver(sys.argv[1])
else:
    receiver = sacn.sACNreceiver()
receiver.start()  # start the receiving thread

# define a callback function
@receiver.listen_on('universe', universe=UNIVERSE)  # listens on universe 1
def callback(packet):  # packet type: sacn.DataPacket
    print("Recieved DMX data: ==> ",packet.dmxData[(first_channel):(first_channel + numb_lights*channels_per_light)])
    for idx, light in enumerate(lights):
        # color is [Hue, Saturation, Brightness, Kelvin], duration in ms
        start, end = dmx_channels(idx)
        rgb = packet.dmxData[start:end]
        hsb = [int(x * scale_up) for x in colorsys.rgb_to_hsv(*[c / max_rgb_per_color for c in rgb])]
        if sum(rgb) > override_max_bright_total_rgb:
            adjust_perc =  (sum(rgb) - override_max_bright_total_rgb) * override_scaling
            bright_adj = max_bright + ((max_bright_override - max_bright) * adjust_perc)
        else:
            adjust_perc = 0
            bright_adj = max_bright
        hsb[2] = round(hsb[2] * bright_adj) # Adjust the brightness
        color = tuple(hsb) + (3500,) # Add kelvin
        print(f"{light.label:20} set to {color} from RGB{rgb} - Adjust % {adjust_perc} bright_adj {bright_adj}")
        light.set_color(color, 0, True)

# optional: if you want to use multicast use this function with the universe as parameter
receiver.join_multicast(UNIVERSE)

def cleanup():
    receiver.stop()
    print("Restoring original colors and power to all lights...")
    for light in lights:
        light.set_color(original_colors[light])
        light.set_power(original_powers[light])

def signal_handler(sig, frame):
    print("You pressed Ctrl+C!")
    cleanup()
    sys.exit()

signal.signal(signal.SIGINT, signal_handler)
print("Press Ctrl+C to quit")

[mclarkk]

Thanks for the details about the network setup!

Agreed that discovery doesn't work so well. I wrote this library about 10 years ago when I had less experience. I would certainly write it differently now, but I don't have the free time to fix it, and I've moved away from LIFX products over the last decade. I added another maintainer to help a couple years ago, but I think we're all running into this same issue of having other things to do.

Manual light creation would be the way to go if discovery isn't working consistently. What did you see happen (or not happen) when you were using manual light creation? Can you write a test script that manually connects to just one or two lights without any of the DMX stuff and report back whether that works or fails? It is difficult for me to tell from the scripts shared so far whether this is a lifxlan problem (besides discovery) or if the issues are with the way the script integrates with DMX.

[mclarkk]

(More of a note to myself: I'm poking around the discovery code, and I assume the issue is that the default timeout is set to one second and the default number of attempts is one. With 47 lights sending messages over UDP, presumably you need a longer timeout and/or more attempts during discovery. One option for a good update for lifxlan would be to surface those as optional parameters to get_lights() so users can configure it, and/or incorporate the fact that we know there are 47 lights to be found, both to increase the parameters and to cut out early once we've found all the devices.)