asyncio.create_task() breaks the UDP packet transmission

[sergimn]

Using asyncio.create_task(bulb.turn_on()) makes the library raise a WizLightConnectionError. Expected behavior is that the UDP packet would be relayed anyways and I could make a similar call to turn on an indefinite amount of lightbulbs, with the help of a for loop, for example. Note that tcpdump shows that no packet is sent

Example code to reproduce the error:

import asyncio
from pywizlight import  wizlight

async def test():
    bulb = wizlight("192.168.0.10")
    asyncio.create_task(bulb.turn_on())

if __name__ == '__main__':
    asyncio.run(test())

[sbidy]

Please have a look into the docs and example in the readme. You have to pass a PilotBuilder() with the turn_on() call and await for it. Creating a separate task is (currently) not possible.

import asyncio
from pywizlight import  wizlight, PilotBuilder

async def test():
    bulb = wizlight("192.168.0.10")
    await bulb.turn_on(PilotBuilder())

if __name__ == '__main__':
    asyncio.run(test())

[sergimn]

Hi :)

I was looking for something like create_task() as I have multiple bulbs and my idea was that I could control them in different threads. Moreover, the reason I don't use asyncio.gather() is that the number of bulbs I'm going to use is not a fixed amount, and asyncio.gather() demands a fixed amount of parameters.

If you can guide me in how to implement it I may submit a PR in the future fixing this

[sbidy]

Mh that is difficult to use an loop without an fixed number of entries. Maybe asyncio-queue can help you to add the request to the an queue.

[sbidy]

Dose the queue the job? 😉

[sergimn]

At this moment I'm not able to test as I don't have the bulb with me. I'll try to look into it next month

[eibanez]

I am not sure if this would help, but you could run the code below to activate a number of light bulbs, determine by a list of their IPs.

from pywizlight import  wizlight, PilotBuilder

async def test():
    ips = [...]
    bulbs = [wizlight(ip) for ip in ips]
    await asyncio.gather(*[b.turn_on(PilotBuilder()) for b in bulbs])

if __name__ == '__main__':
    asyncio.run(test())

[bdraco]

There is a lock in place now. Hopefully this works ok with the latest version.

[bdraco]

This can be closed. The original code did not await the task so the event loop would be terminated before it could complete.