Gateway

It is the home instance gateway to the Lifx bus. Connects triggers and commands to the Lifx bus.

class lifx_plugin.gateway.Gateway(client: Type[lifx_plugin.client.Client], address: str = '0.0.0.0', port: int = 56700)

>>> import asyncio
>>> import lifx_plugin

>>> loop = asyncio.get_event_loop()

>>> description = {"type": "lifx",
...                 "name": "SetColor",
...                 "fields": {},
...                 "addresses": [["127.0.0.1", 56701]]}
>>> command = lifx_plugin.Command(description)
>>> trigger = lifx_plugin.Trigger(description)
>>> gateway = Gateway(lifx_plugin.Client, "127.0.0.1", 56700)
>>> gateway.associate_commands([command])
>>> gateway.associate_triggers([trigger])
>>> async def a_task(msgs):
...     print(msgs)
...     loop.stop()
>>> t = loop.create_task(gateway.run([a_task]))

>>> description = {"type": "lifx",
...                 "name": "State",
...                 "fields": {},
...                 "addresses": [["127.0.0.1", 56700]]}
>>> command = lifx_plugin.Command(description)
>>> lifx_lamp = Gateway(lifx_plugin.Client, "127.0.0.1", 56701)
>>> lifx_lamp.associate_commands([command])
>>> t = loop.create_task(lifx_lamp.run([]))
>>> msgs = command.execute()
>>> t = loop.create_task(lifx_lamp.writer(msgs))
>>> loop.run_forever()
Trigger State {power: 0, hue: 0, saturation: 0, brightness: 0, kelvin: 0, rgb: (0, 0, 0), label: } from [127.0.0.1:56701]

PROTOCOL = 'lifx'

async disconnect()

associate_commands(descriptions: List[lifx_plugin.message.Description])

Connect the Protocol Commands with the Gateway, if needed.

Only connected commands should go through the Gateway to the bus.

Parameters

descriptions – a protocol message description

Returns

an association object, if any

associate_triggers(descriptions: List[lifx_plugin.message.Description])

Connect the Protocol Triggers with the Gateway, if needed.

Only connected triggers should go through the Gateway from the bus.

Parameters

descriptions – a protocol message description

Returns

an association object, if any

async run(other_tasks: List[Callable])

Create an asynchronous task capable of receiving messages from the bus.

Every received message becomes a Protocol Trigger.

Every created Protocol Trigger goes through given tasks to be processed.

Parameters

tasks – a function capable of process a Protocol Trigger

async writer(msgs: List[lifx.lan.msg.Msg], *args)

A function capable of sending protocol messages to the bus.

Parameters

• msgs – a protocol messages

• args –

static make_trigger(msg: lifx.lan.msg.Msg)

Make a Protocol Trigger from a protocol message.

Parameters

msg – a protocol message

Returns

a Protocol Trigger