The Gatherer Interface

A common task is to gather information from devices. This can often require combining multiple points of information. For example to determine if the device supports extended multizone messages you need both the product id and firmware version of the device.

To get and combine this information in a way that streams that information back to you using the sender API is a little tricky.

To solve this problem, Photons provides the gatherer that lets you ask for plans of information to be resolved.

For example:

from photons_control.planner import Skip


async def my_action(target, reference):
    async with target.session() as sender:
        plans = sender.make_plans("capability", "firmware_effects")

        async for serial, name, info in sender.gatherer.gather(plans, reference):
            if name == "capability":
                print(f"{serial} is a {info['cap'].product.name}")

            elif name == "firmware_effects":
                if info is Skip:
                    print(f"{serial} doesn't support firmware effects")
                else:
                    print(f"{serial} is running the effect: {info['type']}")

In this example each device will get one GetVersion, GetHostFirmware and for those that support firmware effects, either a GetMultizoneEffect or GetTileEffect and all the information from those is presented to you in a useful way. Note that both the "capability" and "firmware_effects" plans require version information, but the gatherer makes it so we only ask the device for this information once.

By using gather we will get each plan result as they come in. The gatherer also gives you gather_per_serial and gather_all.

All the gather methods take in plans, reference and **kwargs where kwargs are the same keyword arguments you would give the sender.

The gather methods

There are three methods on the sender.gatherer that you would use:

class photons_control.planner.gatherer.Gatherer

async Gatherer.gather(plans, reference, error_catcher=None, **kwargs)

This is an async generator that yields tuples of (serial, label, info) where serial is the serial of the device, label is the label of the plan, and info is the result of executing that plan.

The error handling of this function is the same as the async generator behaviour in sender API.

async Gatherer.gather_per_serial(plans, reference, **kwargs)

This is an async generator that yields tuples of (serial, complete, info) where serial is the serial of the device, complete is a boolean that indicates if all the plans resolved for this device; and info is the result from all the plans.

info will look like {<label>: <result of plan>}

The error handling of this function is the same as the async generator behaviour in sender API.

async Gatherer.gather_all(plans, reference, **kwargs)

This is a coroutine that returns a single dictionary that looks like {serial: (completed, info)}.

The error handling of this function is the same as the synchronous behaviour in sender API.

This is a shortcut for getting all (serial, completed, info) information from the gather_per_serial method.

Using Plans

There are a number of plans that comes with Photons by default. You can either use these by their name when you use make_plans function or you can use your own label with something like make_plans(thing=MyThingPlan()) and then the result from that plan will have the label thing.

For example:

plans = sender.make_plans("capability", "state", other=MyOtherPlan())

Is the same as saying:

from photons_control.planner.plans import CapabilityPlan, StatePlan

plans = sender.make_plans(capability=CapabilityPlan(), state=StatePlan, other=MyOtherPlan())

The first form is convenient, but the second form gives you more control on what label the plan is given and lets you override the refresh value of the plan which says how long it’ll take for the data it cares about to refresh and be asked for again.

Photons comes with some default plans for you to use:

address
capability
chain
colors
firmware
firmware_effects
hev_config
hev_status
label
parts
parts_and_colors
power
presence
state
version
zones
PacketPlan
SetZonesPlan

`"address"`

class photons_control.planner.plans.AddressPlan(refresh=10): Return the (ip, port) for this device

`"capability"`

class photons_control.planner.plans.CapabilityPlan(refresh=10)

Return capability information for this device:

{
  "cap": <capability object with filled out firmware version>,
  "product": <product object>,
  "firmware": <object with build/version_major/version_minor attributes>,
  "state_version": <the StateVersion packet we received from the device>
}

The capability and product objects come from the registry.

`"chain"`

class photons_control.planner.plans.ChainPlan(refresh=1)

This plan will return chain information about the device in a dictionary.

It is recommended you instead use the “parts” or “parts_and_colors” plans instead.

The result includes:

chain: A list of photons_messages.fields.Tile objects.
width: The maximum width of the chain items.
orientation: A dictionary of chain index to orientation.
reorient: A function that takes in chain index and list of colors. It will return those colors but reoriented to appear upright on the device.
reverse_orient: A function that takes in the chain index and a list of colors. It is used to take the colors from the device and rotate them to appear upright in your program
coords_and_sizes: A list of ((x, y), (width, height)) for each item in the chain. This is taken from the TileMessages.StateDeviceChain packet.
random_orientations: These are used if you give randomize=True to orient or reverse_orient. You can change this list by using __import__("random").shuffle(random_orientations)

For strips and bulbs we will return a single chain item that is orientated right side up. For strips, the width of this single item is the number of zones in the device.

`"colors"`

class photons_control.planner.plans.ColorsPlan(refresh=1)

Return [[hsbk, ...]] for all the items in the chain of the device.

So for a bulb you’ll get [[<hsbk>]].

For a Strip or candle you’ll get [[<hsbk>, <hsbk>, ...]]

And for a tile you’ll get [[<hsbk>, <hsbk>, ...], [<hsbk>, <hsbk>, ...]]

Where <hsbk> is a photons_messages.fields.Color object.

`"firmware"`

class photons_control.planner.plans.FirmwarePlan(refresh=10)

Return in a dictionary

build - The build timestamp of this firmware
version_major - the major component of the firmware version
version_minor - the minor component of the firmware version

`"firmware_effects"`

class photons_control.planner.plans.FirmwareEffectsPlan(refresh=1)

Return {"type": <enum>, "options": {...}}` for each device where strips return multizone effect data and tiles return tile effect data.

Returns Skip for devices that don’t have firmware effects

`"hev_config"`

class photons_control.planner.plans.HevConfigPlan(refresh=10)

Returns the default HEV configuration for the device:

{
    "indication": <boolean>,
    "duration_s": <time in seconds>
}

indication: whether a short flashing indication occurs when the cycle ends
duration_s: the total time in seconds for the current cycle

`"hev_status"`

class photons_control.planner.plans.HevStatusPlan(refresh=1)

Returns the current and previous HEV cycle run info for a device. The duration, remaining and power_off fields are only included if there is an HEV cycle currently running:

 {
     current: {
         active: <boolean>,
         duration_s: <total time in seconds>
         remaining: <time left in seconds>
         power_off: <boolean>
     },
     last: {
         result: <result of last cycle>
     }
}

current:
- active: (boolean) true if the device is currently in a cycle
- duration_s: the total time in seconds for the current cycle
- remaining: the remaining time in seconds for the current cycle
- power_off: (boolean) true if the device power itself off when the cycle completes
last:
- result: either SUCCESS, BUSY or INTERRUPTED BY <source>

`"label"`

class photons_control.planner.plans.LabelPlan(refresh=5): Return the label of this device

`"parts"`

class photons_control.planner.plans.PartsPlan(refresh=1): Return a list of photons_canvas Part objects for this device.

`"parts_and_colors"`

class photons_control.planner.plans.PartsAndColorsPlan(refresh=1)

Return a list of photons_canvas Part objects for this device.

Note that each part will have original_colors set to the current colors on the device.

`"power"`

class photons_control.planner.plans.PowerPlan(refresh=1)

Return in a dictionary:

level: The current power level on the device
on: False if the level is 0, otherwise True

`"presence"`

class photons_control.planner.plans.PresencePlan(refresh=10)

Just return True.

This can be used with other plans to make sure that this serial is returned by gather if no other plans return results

`"state"`

class photons_control.planner.plans.StatePlan(refresh=1)

Return in a dictionary:

hue
saturation
brightness
kelvin
label
power

`"version"`

class photons_control.planner.plans.VersionPlan(refresh=10)

Return in a dictionary:

vendor: The vendor id of the device.
product: The product id of the device.

`"zones"`

class photons_control.planner.plans.ZonesPlan(refresh=1)

Return a list of [(index, hsbk), ...] for this device.

This plan will take into account if the device supports extended multizone or not

PacketPlan(…)

class photons_control.planner.plans.PacketPlan(self, sender_pkt, receiver_kls)

Takes in a packet to send and a packet class to expect.

If we successfully get the correct type of packet, then we return that packet.

from photons_control.planner import PacketPlan
from photons_messages import LightMessages

plans = sender.make_plans(
    infrared=PacketPlan(LightMessages.GetInfrared(), LightMessages.StateInfrared)
)
async for serial, label, info in sender.gatherer.gather(plans):
    if label == "infrared":
        # info will be a StateInfrared packet

SetZonesPlan(…)

class photons_control.multizone.SetZonesPlan(self, colors, zone_index=0, duration=1, overrides=None, refresh=True)

Return messages used to apply a color range to multizone devices

Takes in:

colors - [[color_specifier, length], ...]: For example, [["red", 1], ["blue", 3], ["hue:100 saturation:0.5", 5]] will set one zone to red, followed by 3 zones to blue, followed by 5 zones to half saturation green.
zone_index - default 0: An integer representing where on the device to start the colors
duration - default 1: Time it takes to apply.
overrides - default None: A dictionary containing hue, saturation, brightness and kelvin for overriding colors with

For devices that don’t have multizone capabilities, the info will be a Skip otherwise, you’ll get the appropriate messages to then send to the device to apply the change.

Usage is:

from photons_control.planner import Skip

plans = {"set_zones": SetZonesPlan(colors)}

async with target.session() as sender:
    async for serial, name, messages in sender.gatherer.gather(plans, reference):
        if name == "set_zones" and messages is not Skip:
            await sender(messages, serial)

Note that this example code will do one device at a time, If you want to change multiple devices at the same time then use photons_control.multizone.SetZones message instead.

Making your own Plans

To make your own Plan, you need to inherit from the photons_control.planner.Plan class and fill out some methods.

The simplest plan you can make is:

from photons_control.planner import Plan, a_plan
from photons_messages import DeviceMessages


@a_plan("responds")
class RespondsPlan(Plan):
    """
    Return whether this device responds to messages
    """

    messages = [DeviceMessages.EchoRequest(echoing=b"can_i_has_response")]

    class Instance(Plan.Instance):
        def process(self, pkt):
            if pkt | DeviceMessages.EchoResponse:
                if pkt.echoing == b"can_i_has_response":
                    return True

        async def info(self):
            return True

Here we say the plan should send out a single EchoRequest for this device. We then have an Instance class that inherits from Plan.Instance and one of these is made for every device the Plan is operating on. There are two main methods on this: process and info.

The process method takes in every packet that has been received from the device, even if this plan didn’t send that packet.

You can store whatever information you want on self and when you have received enough information you return True from this method. Once that is done, the info method is called to return something useful from this plan.

In our example above, the info from the plan is a boolean True if the device responded to our echo request. If we never get that response back then this plan will never resolve and not given back in the output from the gather methods.

You can also depend on other plans:

from photons_control.planner.plans import CapabilityPlan
from photons_control.planner import Plan, a_plan, Skip
from photons_messages import DeviceMessages
from photons_products import Family


@a_plan("labels_from_lcm1")
class LabelsFromLCM1(Plan):
    @property
    def dependant_info(kls):
        return {"c": CapabilityPlan()}

    class Instance(Plan.Instance):
        @property
        def is_lcm1(self):
            return self.deps["c"]["cap"].product.family is Family.LCM1

        @property
        def messages(self):
            if self.is_lcm1:
                return [DeviceMessages.GetLabel()]
            return Skip

        def process(self, pkt):
            if pkt | DeviceMessages.StateLabel:
                self.label = pkt.label
                return True

        async def info(self):
            return self.label

This plan will return the label for any LCM1 devices it finds. If the device is not an LCM1 then the info for this plan will be a Skip object.

The gatherer will make sure all dependencies for the plan are resolved before this plan is used. If that dependency has any dependencies, then those are resolve first, and so on.

Note

the a_plan decorator is optional and is only registering this plan with a label you can use as a positional argument in the make_plans function.

You can look at the source code for examples of plans.

There’s quite a few features on the Plan object:

class photons_control.planner.Plan(refresh=10, **kwargs)

Base class for plans. A plan is an object that specifies what messages to send to a device, How to process any message that the device sends back and how to create a final artifact that represents the value from this plan.

Usage looks like:

class MyPlan(Plan):
    messages = [DeviceMessages.GetPower()]

    class Instance(Plan.Instance):
        def process(self, pkt):
            if pkt | DeviceMessages.StatePower:
                self.level = pkt.level
                return True

        async def info(self):
            return {"level": self.level}

There are some properties on the Plan itself:

messages - Default None: a list of messages to send to the device. If you want to choose based on the device itself, then define messages on the Instance where you have access to the deps and serial.
dependant_info - Default None: An optional dictionary of key to Plan class that is used to get dependency information for this plan. For example if you say dependant_info = {"c": CapabilityPlan()} then the plan will only be executed once we have information from the CapabilityPlan and the Instance will have self.deps = {"c": <info from capability plan>} for use.
default_refresh - Default 10: Either True if you never want to re-use results from this plan. False if you never want to remove the results from this plan, or an integer representing the number of seconds before the cache expires. If the cache expires then the messages defined by this plan are re-sent to the device. Note that this is overridden if you specify refresh when you instantiate the plan.
setup - Method: Called by __init__ with all positional and keyword arguments used to instantiate the plan except refresh. Note that before setup is called, self.refresh will be set on the class as either the refresh keyword argument if provided, or the default_refresh on the plan
Instance - Class - Should inherit photons_control.planner.plans.Plan.Instance: The logic for processing packets from the device and getting the final information for the plan.

The Instance class will be instantiated per device that the planner is looking at. It is also possible to define messages on the Instance. If you want to have logic for determining what messages to send, then just make messages a method with an @property decorator.

If you define messages on the instance you have access to:

self.parent – the plan instance used to create this device Instance
self.deps – Dependency information as defined by dependant_info on the parent
self.serial – the serial for this device
Anything else on the instance defined in the setup hook

The Instance has the following hooks and properties:

Messages - Default None: A list of messages to send to the device. If this is not defined, then messages on the parent is used. If this is Skip then the plan is not executed for this device.
finished_after_no_more_messages - default False: This plan should be considered done if it already isn’t considered done and we don’t have any more messages from devices.
setup - Method: Any setup specific to this device. You have self.deps, self.parent and self.serial at this point.
refresh - property: Defaults to self.parent.refresh and is the refresh for this instance.
key - Method: Defaults to self.parent.__class__ and is used to represent this plan, so that future executions of this plan can reuse the final information for this serial without sending more messages.
process - Method: Does nothing by default and is used to process all messages that the device sends back, not just replies to the messages asked for by the plan. You must return True from this method when you have received enough information. Once we return True then the info method will be called to get the final result for this plan. Note that if the messages on this plan is the NoMessages class then this method will never be called.
info - Async method - Must be overridden: This is called when the plan is considered done for this device and returns whatever information you want.

The Gatherer Interface

The gather methods

Using Plans

"address"

"capability"

"chain"

"colors"

"firmware"

"firmware_effects"

"hev_config"

"hev_status"

"label"

"parts"

"parts_and_colors"

"power"

"presence"

"state"

"version"

"zones"

PacketPlan(…)

SetZonesPlan(…)

Making your own Plans

`"address"`

`"capability"`

`"chain"`

`"colors"`

`"firmware"`

`"firmware_effects"`

`"hev_config"`

`"hev_status"`

`"label"`

`"parts"`

`"parts_and_colors"`

`"power"`

`"presence"`

`"state"`

`"version"`

`"zones"`