Gateway

The Automate Home Gateway for the Sonos protocol.

Connects triggers and commands to the Sonos protocol through the SoCo library.

class soco_plugin.gateway.Gateway

PROTOCOL = 'soco'

SEND_MESSAGE_TIMEOUT = 1

POLL_TIMEOUT = 0.05

async unlock_events(player)

async lock_events(player)

is_locked(player)

async disconnect() → None

async associate_commands(descriptions: Iterable[soco_plugin.Command]) → None

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

async associate_triggers(descriptions: Iterable[soco_plugin.Trigger]) → None

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

build_msg()

build_msgs_from_bus(player: soco.core.SoCo, event) → Iterable[soco_plugin.Description]

static get_action(player: soco.core.SoCo, msg: soco_plugin.message.Command) → Callable

>>> import home
>>> import soco_plugin
>>> class Player:
...     def play(self):
...         print("play")
...     def pause(self):
...         print("pause")
...     def stop(self):
...         print("stop")
...     @property
...     def volume(self):
...         return 1
...     @property
...     def play_mode(self):
...         return "booo"
...     @volume.setter
...     def volume(self, value):
...         print("volume is {}".format(value))
...     @play_mode.setter
...     def play_mode(self, mode):
...         print("play mode is {}".format(mode))
...     def get_sonos_playlist_by_attr(self, attr, title):
...         print ("playlist title is {}".format(title))
...         return {"uri": "a uri"}
...     def clear_queue(self):
...         pass
...     def add_uri_to_queue(self, uri):
...         pass
...     def play_from_queue(self, index):
...         pass
...     def ramp_to_volume(self, volume, ramp_type):
...         print("ramp to {} {}".format(volume, ramp_type))

>>> old_state = home.appliance.sound.player.state.off.State()
>>> new_state = old_state.next(home.appliance.sound.player.event.forced.Event.On)
>>> new_state = new_state.next(home.event.sleepiness.Event.Awake)
>>> msgs = soco_plugin.command.play.Command.make(["Bath"]).make_msgs_from(old_state, new_state)
>>> action = soco_plugin.Gateway.get_action(Player(), msgs[0])
>>> action()
play
>>> msgs = soco_plugin.command.volume.absolute.Command.make(["Bath"]).make_msgs_from(old_state, new_state)
>>> action = soco_plugin.Gateway.get_action(Player(), msgs[0])
>>> action()
volume is 20
>>> msgs = soco_plugin.command.volume.relative.Command.make(["Bath"]).make_msgs_from(old_state, new_state)
>>> action = soco_plugin.Gateway.get_action(Player(), msgs[0])
>>> action()
volume is 11
>>> msgs = soco_plugin.command.playlist.Command.make(["Bath"]).make_msgs_from(old_state, new_state)
>>> action = soco_plugin.Gateway.get_action(Player(), msgs[0])
>>> action()
playlist title is Common
>>> old_state = home.appliance.sound.player.state.off.State()
>>> old_state = old_state.next(home.event.presence.Event.On)
>>> old_state = old_state.next(home.event.sleepiness.Event.Asleep)
>>> new_state = old_state.next(home.event.sleepiness.Event.Awake)
>>> msgs = soco_plugin.command.volume.ramp.Command.make(["Bath"]).make_msgs_from(old_state, new_state)
>>> action = soco_plugin.Gateway.get_action(Player(), msgs[0])
>>> action()
ramp to 30 SLEEP_TIMER_RAMP_TYPE
>>> msgs = soco_plugin.command.mode.Command.make(["Bath"]).make_msgs_from(old_state, new_state)
>>> action = soco_plugin.Gateway.get_action(Player(), msgs[0])
>>> action()
play mode is SHUFFLE

Parameters

msg – a command message to be executed
player – a player where the command will be applied

Returns

a callable to execute

async send_msg(msg: soco_plugin.message.Command, player: soco.core.SoCo) → None

get_response_type(msg: soco_plugin.message.Command)

async writer(msgs: Iterable[soco_plugin.message.Command], *args) → None

A function capable of sending protocol messages to the bus.

Parameters

msgs – a protocol messages
args –

static make_trigger(msg: soco_plugin.Description) → soco_plugin.Trigger

Make a Protocol Trigger from a protocol message.

Parameters: msg – a protocol message
Returns: a Protocol Trigger

async run(other_tasks: Iterable[Callable]) → None

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