Special Message objects

Using the sender we can send individual packets but sometimes we want to have more control over the order and timing of packets that is otherwise difficult and tedious to do.

For this purpose, you can provide the sender an object that makes it easy to make these kind of decisions.

For example, an object that toggles the power of lights:

from photons_messages import DeviceMessages, LightMessages
from photons_control.script import FromGenerator


async def toggle(reference, sender, **kwargs):
    get_power = DeviceMessages.GetPower()

    async for pkt in sender(get_power, reference, **kwargs):
        if pkt | DeviceMessages.StatePower:
            if pkt.level == 0:
                yield LightMessages.SetLightPower(
                    level=65535, res_required=False, target=pkt.serial
                )

            else:
                yield LightMessages.SetLightPower(
                    level=0, res_required=False, target=pkt.serial
                )


async def my_action(target, reference):
    async with target.session() as sender:
        ToggleLights = FromGenerator(toggle)
        await sender(ToggleLights, reference)

Here we’re using the FromGenerator helper to turn an async generator into a message object. What this means is that every message that is yielded from will be sent to devices. See below for more details about how this mechanism works.

There exists a few message objects you can use straight away. For example, if I wanted to repeatedly send certain messages forever I could do something like:

from photons_messages import DeviceMessages
from photons_control.script import Repeater


async def my_action(target, reference):
    get_power = DeviceMessages.GetPower()
    get_group = DeviceMessages.GetGroup()

    def errors(e):
        print(e)

    msg = Repeater([get_power, get_group], min_loop_time=5)

    async for pkt in target.send(msg, reference, error_catcher=errors, message_timeout=3):
        if pkt | DeviceMessages.StatePower:
            power_state = "off" if pkt.level == 0 else "on"
            print(f"{pkt.serial} is {power_state}")

        elif pkt | DeviceMessages.StateGroup:
            print(f"{pkt.serial} is in the {pkt.label} group")

Here our msg object will keep asking the devices for power and group until the program is stopped. The Repeater takes in a couple options, and here we’re using the min_loop_time time to say if we get replies before it’s been 5 seconds, then wait the remaining time before sending the messages again.

You can even combine special messages. So say I want to keep toggling my lights forever, I can use our ToggleLights msg above with the Repeater:

from photons_control.script import Repeater, FromGenerator


async def toggle(reference, sender, **kwargs):
    ....

async def my_action(target, reference):
    def errors(e):
        print(e)

    ToggleLights = FromGenerator(toggle)
    msg = Repeater(ToggleLights, min_loop_time=5)
    await target.send(msg, reference, error_catcher=errors, message_timeout=3)

You can even do this without making your own toggle function!

from photons_control.transform import PowerToggle
from photons_control.script import Repeater


async def my_action(target, reference):
    def errors(e):
        print(e)

    msg = Repeater(PowerToggle(), min_loop_time=5)
    await target.send(msg, reference, error_catcher=errors, message_timeout=3)

Existing message objects

photons_control.script.Repeater(msg, min_loop_time=30, on_done_loop=None)

This will send the provided msg in an infinite loop

For example:

msg1 = MessageOne()
msg2 = MessageTwo()

reference = ["d073d500000", "d073d500001"]

def error_catcher(e):
    print(e)

async with target.session() as sender:
    pipeline = Pipeline(msg1, msg2, spread=1)
    repeater = Repeater(pipeline, min_loop_time=20)
    async for result in sender(repeater, reference, error_catcher=error_catcher):
        ....

Is equivalent to:

async with target.session() as sender:
    while True:
        start = time.time()
        async for result in sender(pipeline, reference):
            ...
        await asyncio.sleep(20 - (time.time() - start))

Note that if references is a SpecialReference then we call reset on it after every loop.

Also it is highly recommended that error_catcher is a callable that takes each error as they happen.

Repeater takes in the following keyword arguments:

min_loop_time

The minimum amount of time a loop should take, in seconds.

So if min_loop_time is 30 and the loop takes 10 seconds, then we’ll wait 20 seconds before going again

on_done_loop

An async callable that is called with no arguments when a loop completes (before any sleep from min_loop_time)

Note that if you raise Repeater.Stop() in this function then the Repeater will stop.

photons_control.script.Pipeline(*messages, spread=0, short_circuit_on_error=False, synchronized=False)

This allows you to send messages in order, so that each message isn’t sent until the previous message gets a reply or times out.

For example:

msg1 = MessageOne()
msg2 = MessageTwo()

reference1 = "d073d500000"
reference2 = "d073d500001"

async with target.session() as sender:
    async for result in sender(Pipeline(msg1, msg2), [reference1, reference2]):
        ....

Is equivalent to:

async with target.session() as sender:
    async for result in sender(msg1, [reference1]):
        ...
    async for result in sender(msg2, [reference1]):
        ...
    async for result in sender(msg1, [reference2]):
        ...
    async for result in sender(msg2, [reference2]):
        ...

This also takes in the following keyword arguments:

spread

Specify the minimum time between sending each message

short_circuit_on_error

Stop trying to send messages to a device if we encounter an error.

This will only effect sending messages to the device that had an error

synchronized

If this is set to False then each bulb gets it’s own coroutine and so messages will go at different times depending on how slow/fast each bulb is.

If this is set to True then we wait on the slowest bulb before going to the next message.

photons_control.script.ForCapability(**by_cap)

Send messages based on capability of the device.

For example:

ForCapability(hev=LightMessages.GetHevCycle())

Will send that message only to devices that have the hev capability. This generator will use the capability plan to find devices with that capability.

You can also specify a group of capabilities:

ForCapability(**{"ir,buttons": LightMessages.SetPower(level=0)})

Will turn off any device that has infrared or buttons.

photons_control.transform.PowerToggle(duration=1, group=False, **kwargs)

Returns a valid message that will toggle the power of devices used against it.

If group is True, then we return a PowerToggleGroup message and treat all lights in the reference as a group.

For example:

await target.send(PowerToggle(), ["d073d5000001", "d073d5000001"])

photons_control.transform.PowerToggleGroup(duration=1, **kwargs)

Returns a valid message that will toggle the power of devices used against it.

This takes into account the whole group of lights so if any light is turned on then all lights are turned off, otherwise they are all turned on

For example:

await target.send(PowerToggle(group=True), ["d073d5000001", "d073d5000001"])

photons_control.multizone.SetZones(colors, power_on=True, reference=None, **options)

Set colors on all found multizone devices. This will use the extended multizone messages if they are supported by the device to increase reliability and speed of application.

Usage looks like:

msg = SetZones([["red", 10], ["blue", 10]], zone_index=1, duration=1)
await target.send(msg, reference)

By default the devices will be powered on. If you don’t want this to happen then pass in power_on=False

If you want to target a particular device or devices, pass in a reference.

The options to this helper include:

colors - [[color_specifier, length], …]: For example, [[“red”, 1], [“blue”, 3], [“hue:100 saturation:0.5”, 5]]
zone_index - default 0: An integer representing where on the device to start the colors
duration - default 1: Application duration
overrides - default None: A dictionary containing hue, saturation, brightness and kelvin for overriding colors with

photons_control.multizone.SetZonesEffect(effect, power_on=True, power_on_duration=1, reference=None, **options)

Set an effect on your multizone devices

Where effect is one of the available effect types:

OFF: Turn the animation off
MOVE: A moving animation

Options include:

speed: duration in seconds to complete one cycle
duration: in seconds or specify 0 (the default) to run until manually stopped
direction: either “left” or “right” (default: “right”)

Usage looks like:

msg = SetZonesEffect("MOVE", speed=1, duration=10, direction="left")
await target.send(msg, reference)

By default the devices will be powered on. If you don’t want this to happen then pass in power_on=False

If you want to target a particular device or devices, pass in reference.

photons_control.tile.SetTileEffect(effect, power_on=True, power_on_duration=1, reference=None, **options)

Set a firmware effect on your Tiles, Candle, Spot, Path or Ceiling

Where effect is one of the available effect types:

OFF: Turn the animation off
FLAME: A flame effect
MORPH: A Morph effect
SKY: A sky effect only supported on LIFX Ceiling, running firmware 4.4 or higher

Options include:

speed
duration
sky_type (SKY effect)
cloud_saturation_min (SKY effect)
cloud_saturation_max (SKY effect)
palette

Usage looks like:

msg = SetTileEffect("MORPH", palette=["red", "blue", "green"])
await target.send(msg, reference)

By default the devices will be powered on. If you don’t want this to happen then pass in power_on=False.

If you want to target a particular device or devices, pass in reference.

class photons_control.transform.Transformer

This is responsible for creating the messages to send to a device for a transformation

msg = Transformer.using({"power": "on", "color": "red"}, keep_brightness=False)
async for info in target.send(msg, references):
    ...

If there are no color related attributes specified then we just generate a message to set the power on or off as specified.

If we are turning the light on and have color options then we will first ask the device what it’s current state is. Lights that are off will be set to brightness 0 on the new color and on, and then the brightness will be changed to match the end result.

For the color options we use ColourParser to create the SetWaveformOptional message that is needed to change the device. This means that we support an effect option for setting different waveforms.

If keep_brightness=True then we do not change the brightness of the device despite any brightness options in the color options.

If transition_color=True then we do not change the color of the device prior to turning it on, so that it transitions with the brightness.

class photons_canvas.theme.ApplyTheme

Apply a theme to your devices.

Usage looks like:

options = {"colors": [<color>, <color>, ...]}
await target.send(ApplyTheme.msg(options))

The options available are:

colors

A list of color specifiers

duration

How long the transition takes. Defaults to 1 second

power_on

Whether to also power on devices. Defaults to true

overrides

A dictionary of {"hue": 0-360, "saturation": 0-1, "brightness": 0-1, "kelvin": 2500-9000}

Where each property is optional and will override any color set in the theme.

Making your own message objects

To make your own message objects you use the FromGenerater helper mentioned above, or the related helper photons_control.script.FromGeneratorPerSerial.

FromGeneratorPerSerial works exactly like FromGenerater except that the reference passed in will be each individual serial and the messages you yield will automatically be told to send to that serial.

class photons_control.script.FromGenerator(generator, *, reference_override=None, error_catcher_override=None)

FromGenerator let’s you determine what messages to send in an async generator.

For example:

async def gen(reference, sender, **kwargs):
    get_power = DeviceMessages.GetPower()

    for pkt in sender(get_power, reference, **kwargs):
        if pkt | DeviceMessages.StatePower:
            if pkt.level == 0:
                yield DeviceMessages.SetPower(level=65535, target=pkt.serial)
            else:
                yield DeviceMessages.SetPower(level=0, target=pkt.serial)

msg = FromGenerator(gen)
await target.send(msg, "d073d5000001")

The reference your generator will receive will either be the reference from the sender API or the reference_override you supply to the FromGenerator. If reference_override is True then you’ll get the reference from the sender.

If you don’t specify reference_override then the messages you yield must have an explicit target, otherwise they won’t go anyway (unless you supplied broadcast=True to the sender).

If you say reference_override=True then messages that don’t have an explicit target will go to the reference given to you.

Finally, if reference_override is anything else, that’s the reference your generator will get and also the default reference the yielded message will go to.

The **kwargs your generator gets is passed down to you through the sender API. For example:

from photons_control.script import FromGenerator
from photons_messages import DeviceMessages


async def my_action(target, reference):
    errors = []

    async def gen(reference, sender, **kwargs):
        assert kwargs ==  {"message_timeout": 10, "error_catcher": errors}
        yield DeviceMessages.SetPower(level=0)

    msg = FromGenerator(gen, reference_override=True)
    await target.send(msg, "d073d5000001", message_timeout=10, error_catcher=errors)

The return value from yield will be a future that resolves to True if the message(s) were sent without error, and False if they were sent with error.

async def gen(reference, sender, **kwargs):
    f = yield [
        DeviceMessages.SetPower(level=65535),
        LightMessages.SetInfrared(brightness=65535)
    ]

    result = await f
    # result will be True if both of those messages were sent and got
    # a reply. If either timed out, then result will be False

If you want to gather information before yielding messages to send, then you can use the sender like you normally would. The benefit of yielding messages instead of using the sender is that all the messages will be sent in parallel to your devices and you can control the flow of those messages based on the future that yielding returns.