Can we make it work with LIFX bulbs

[sharkwsk]

Hello,

I am trying to see if I can modify this script to control LIFX smart bulb instead ?

//sharkwsk

[chapnickc]

hi @sharkwsk, in theory, yes! It looks like someone wrote a Python API for LIFX. Feel free to create a PR or a fork 💡🎵

[sharkwsk]

Hi @chapnickc, Thank you. I did refer to that API and started modifying your code to see whether I can make it work although I am a beginner to this area :)

I modified the LightController class based on my limited knowledge, this is to control a single LIFX smart light based on SPOTIFY play. Can you guide me if my approach is correct or not ?

Thank you

Modified code so far:

from __future__ import annotations
from . import config
from .AnalysisHelper import AnalysisHelper
from .events import (
        Event,
        EventSongChanged,
        EventStop,
        EventAdjustStartTime,
        RawSpotifyResponse,
        Colors)
# from pywizlight.bulb import wizlight, PilotBuilder
import asyncio
from typing import NoReturn, AsyncIterable, Callable
from bisect import bisect_left
import time
import logging
import random
import colorsys
from tools import gen_packet, get_power_packet
import socket
from lifxlan import BLUE, GREEN, LifxLAN

def get_empty_colors(leds: int) -> Colors:
   return [(255, 90, 0)] * leds

async def _color_generator(leds: int, event_queue: asyncio.Queue[Event]) -> AsyncIterable[Colors]:
    get_current_colors = None
    start_time = 0
    event = EventStop()

    while True:
        while not event_queue.empty():
            event = event_queue.get_nowait()

        if isinstance(event, EventSongChanged):
            start_time = event.start_time
            helper = AnalysisHelper(event.analysis, leds)
            get_current_colors = helper.get_current_colors
        elif isinstance(event, EventAdjustStartTime):
            start_time = event.start_time
        elif isinstance(event, EventStop):
            get_current_colors = None

        if get_current_colors is None:
            yield get_empty_colors(leds)
        else:
            yield get_current_colors(time.time() - start_time)

        print(get_current_colors)

class LightController:
    def __init__(self, event_queue: asyncio.Queue[Event], ip_list: list):
        self.queue = event_queue
        lifx = LifxLAN(1)
        devices = lifx.get_lights()
        self.lights = devices[0]
        self.lights.set_power("on", True)
        #self.lights = [devices[i] for i in range(len(ip_list))]

    async def send_to_device(self, colors: Colors) -> None:
        #if len(colors) != len(self.lights): return
        try:
            ops = [
                #self.lights[i].turn_on(PilotBuilder(rgb=colors[i]))
                self.lights.set_color(colors[i])
                    for i in range(1)
            ]
            #await asyncio.gather(*ops)
        except Exception:
            logging.exception("Something went wrong with LightController")
            await asyncio.sleep(config.CONTROLLER_ERROR_DELAY)

    async def consume(self):
        while True:
            try:
                async for colors in _color_generator(1, self.queue):
                    #print('Color[0]=', colors[0])
                    await self.send_to_device(colors)
            except Exception:
                logging.exception("Something went wrong with LightController")
                await asyncio.sleep(config.CONTROLLER_ERROR_DELAY)

[chapnickc]

Great work, @sharkwsk! Indeed this looks like the right set of changes to interface with LIFX bulbs. One detail--it looks like the LIFX lights expect color as [Hue, Saturation, Brightness, Kelvin], while the existing implementation passes the color as RGB scaled between 0 to 255.

Luckily, the lifx module supports this conversion

So you may need something like this

converted = lifxlan.RGBtoHSBK(colors[i])
self.lights.set_color(converted])

Alternatively you can try the colorsys module from the standard library, and maybe use a fixed value for the temperature/kelvin, since it looks like that's what lifxlan.RGBtoHSBK is doing by default.

import colorsys
converted_hsv = colorsys.rgb_to_hsv(*[n/255. for n in colors[i]]) # note iterable unpacking with '*' prefix
kelvin = 3500
final = [*converted_hsv, kelvin]
self.lights.set_color(converted)

[sharkwsk]

Thank you so much @chapnickc, I managed to get it working. I have some followups which I hope you can asssist

1. Below snip of your code, is it to control multiple Wiz lights ?

           ops = [
               self.lights[i].turn_on(PilotBuilder(rgb=colors[i]))
               for i in range(len(self.lights))
           ]
           await asyncio.gather(*ops)

2. In my case, I am controlling only a single light and I have commented out and set out a single line "set_color" function for the above

   
   from __future__ import annotations
   from . import config
   from .AnalysisHelper import AnalysisHelper
   from .events import (
       Event,
       EventSongChanged,
       EventStop,
       EventAdjustStartTime,
       RawSpotifyResponse,
       Colors)
   import asyncio
   from typing import NoReturn, AsyncIterable, Callable
   from bisect import bisect_left
   import time
   import logging
   import random
   import colorsys
   from tools import gen_packet, get_power_packet, RGBtoHSBK
   import socket
   from lifxlan import BLUE, GREEN, LifxLAN
   import colorsys

def get_empty_colors(leds: int) -> Colors: return [(255, 90, 0)] * leds

async def _color_generator(leds: int, event_queue: asyncio.Queue[Event]) -> AsyncIterable[Colors]: get_current_colors = None start_time = 0 event = EventStop()

while True:
    while not event_queue.empty():
        event = event_queue.get_nowait()

    if isinstance(event, EventSongChanged):
        start_time = event.start_time
        helper = AnalysisHelper(event.analysis, leds)
        get_current_colors = helper.get_current_colors
    elif isinstance(event, EventAdjustStartTime):
        start_time = event.start_time
    elif isinstance(event, EventStop):
        get_current_colors = None

    if get_current_colors is None:
        yield get_empty_colors(leds)
    else:
        yield get_current_colors(time.time() - start_time)

class LightController: def init(self, event_queue: asyncio.Queue[Event], ip_list: list): self.queue = event_queue lifx = LifxLAN(1) devices = lifx.get_lights() self.lights = devices[0] self.lights.set_power("on", True)

self.lights = [devices[i] for i in range(len(ip_list))]

async def send_to_device(self, colors: Colors) -> None:
    #if len(colors) != len(self.lights): return
    try:
        #ops = [
            converted = RGBtoHSBK(colors[0])
            self.lights.set_color(converted)
        #       for i in range(1)
        #    ]
        #await asyncio.gather(*ops)
    except Exception:
        logging.exception("Something went wrong with LightController")
        await asyncio.sleep(config.CONTROLLER_ERROR_DELAY)

async def consume(self):
    while True:
        try:
            async for colors in _color_generator(1, self.queue):
                await self.send_to_device(colors)
        except Exception:
            logging.exception("Something went wrong with LightController")
            await asyncio.sleep(config.CONTROLLER_ERROR_DELAY)

3. During the testing, I do see some level of lag to the played song/beat to how the light switches color. I assume this is something you can expect as its not real time sync, instead sync from the Spotify API data ?