set_red.py

from photons_app.actions import an_action

from photons_control.script import FromGenerator
from photons_messages import LightMessages

from delfick_project.addons import addon_hook
import logging
import asyncio
import time

log = logging.getLogger("red")


@addon_hook(
    extras=[
        ("lifx.photons", "control"),
        ("lifx.photons", "transport"),
        ("lifx.photons", "messages"),
        ("lifx.photons", "device_finder"),
    ]
)
def __lifx__(collector, *args, **kwargs):
    pass


async def iterate(end_after=60, step=5):
    for i in range(0, end_after, step):
        start = time.time()
        yield end_after - i
        diff = step - (time.time() - start)
        if diff > 0:
            await asyncio.sleep(diff)


@an_action(needs_target=True, needs_reference=True, special_reference=True)
async def red(collector, target, reference, **kwargs):
    async def gen(reference, sender, **kwargs):
        async for time_left in iterate():
            print("TIME LEFT", time_left)
            await (yield LightMessages.GetColor())

            await (
                yield LightMessages.SetWaveformOptional(
                    hue=0, saturation=1, cycles=1, period=time_left, res_required=False
                )
            )

    msg = FromGenerator(gen, reference_override=True)

    def error(e):
        log.error(e)

    async for pkt, _, _ in target.script(msg).run_with(reference, error_catcher=error):
        if pkt | LightMessages.LightState:
            print(pkt.serial, "sat: ", pkt.saturation)


if __name__ == "__main__":
    from photons_app.executor import main
    import sys

    main(["lan:red"] + sys.argv[1:])

Note that I'm currently in the process of changing photons API. After my change is complete this script would look like this