Skip to content

Instantly share code, notes, and snippets.

@pcreux
Last active March 16, 2021 21:48
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save pcreux/2f154f7643993ecd7658c4f21c59b2d5 to your computer and use it in GitHub Desktop.
Save pcreux/2f154f7643993ecd7658c4f21c59b2d5 to your computer and use it in GitHub Desktop.
sonospi: Sonos + Ruby + Raspberry PI
require_relative './credentials'
class SonosClient
BASE_URL = 'https://api.ws.sonos.com/control/api/v1'
Error = Class.new(StandardError)
def get(path)
r = HTTP.auth("Bearer #{access_token}")
.get(BASE_URL + path)
raise Error, r.inspect unless r.code == 200
JSON.parse(r.body)
rescue Error => e
refresh_access_token!
raise e # it should work next time
end
def post(path, params = {})
r = HTTP.auth("Bearer #{access_token}")
.post(BASE_URL + path, body: params.to_json)
raise Error, r.inspect unless r.code == 200
JSON.parse(r.body)
end
def access_token
@access_token ||= generate_access_token
end
def refresh_access_token!
@access_token = generate_access_token
end
private def generate_access_token
r = HTTP.basic_auth(user: CLIENT_KEY, pass: CLIENT_SECRET)
.post('https://api.sonos.com/login/v3/oauth/access', form: {
grant_type: 'refresh_token',
refresh_token: REFRESH_TOKEN
})
puts r.body.to_s
creds = JSON.parse(r.body.to_s)
creds.fetch('access_token')
end
end
CLIENT_KEY = 'XXX'
CLIENT_SECRET = 'XXX'
REFRESH_TOKEN = 'XXX'
import RPi.GPIO as GPIO
import sys
GPIO.setmode(GPIO.BCM)
GPIO.setup(int(sys.argv[1]), GPIO.IN, pull_up_down=GPIO.PUD_UP)
require 'http'
require 'cgi'
require 'pp'
require_relative './client'
CLIENT = SonosClient.new
HOUSEHOLD = "Sonos_qZ2xXXX"
Group = Struct.new(:id, :name, :state, :speakers) do
def self.all
@all ||= fetch_all
end
def self.reset!
@all = nil
end
def self.fetch_all
data = CLIENT.get("/households/#{HOUSEHOLD}/groups")
# we cannot trust playbackState here. When a group leaves an existing group, it
# is seen as "playing"
speakers = data.fetch("players").map { |d| Speaker.new(*d.values_at("id", "name")) }
data.fetch("groups").map do |d|
begin
group = Group.new(*d.values_at("id", "name", "playbackState", "playerIds"))
group.speakers = group.speakers.map { |id| speakers.find { |speaker| speaker.id == id } }
group.state = CLIENT.get("/groups/#{group.id}/playback").fetch("playbackState")
group
rescue KeyError => e
pp d
# {"groupStatus":"GROUP_STATUS_GONE"}
# KeyError: key not found: "playbackState"
nil
end
end.compact
end
def self.playing
all.find { |g| g.playing? }
end
def self.first
all.first
end
def self.find(name:)
all.find { |g| g.name == name }
end
def add(speaker)
CLIENT.post("/groups/#{id}/groups/modifyGroupMembers", playerIdsToAdd: [speaker.id])
self.class.reset!
end
def remove(speaker)
CLIENT.post("/groups/#{id}/groups/modifyGroupMembers", playerIdsToRemove: [speaker.id])
self.class.reset!
end
def playing?
state == "PLAYBACK_STATE_PLAYING"
end
end
Speaker = Struct.new(:id, :name) do
def self.all
Group.all.map(&:speakers).flatten.uniq
end
def self.kitchen
find(name: "Kitchen")
end
def self.living_room
find(name: "Living Room")
end
def self.find(name:)
all.find { |s| s.name == name }
end
def toggle
if playing? && alone?
pause!
return
end
if playing? && others?
leave_group!
return
end
if paused? && alone? && !other_group_playing?
play!
return
end
if paused? && alone? && other_group_playing?
join_group!
return
end
if paused? && others?
leave_group!
play!
return
end
end
def leave_group!
puts "Leaving..."
group.remove(self)
end
def join_group!
playing = Group.playing
if playing
puts "Joining..."
playing.add(self)
else
puts "No group playing..."
end
end
def play!
puts "Play!"
CLIENT.post("/groups/#{group.id}/playback/play")
Group.reset!
end
def pause!
puts "Pause!"
CLIENT.post("/groups/#{group.id}/playback/pause")
Group.reset!
end
def playing?
group.playing?
end
def paused?
!playing?
end
def alone?
group.speakers == [self]
end
def others?
!alone?
end
def other_group_playing?
Group.playing && Group.playing != group
end
def group
Group.all.find { |g| g.speakers.include?(self) }
end
end
def k
Speaker.kitchen
end
def l
Speaker.living_room
end
Gpio = Struct.new(:pin) do
def setup
return if File.exists?(base_path)
File.binwrite "/sys/class/gpio/export", pin.to_s
start = Time.now
until File.exists?(base_path + "/direction")
sleep 0.25
if Time.now > start + 5
raise "#{base_path} doesn't exists despite call to export"
end
end
sleep 1
end
def teardown
if File.exists?("/sys/class/gpio/unexport/#{pin}")
File.binwrite("/sys/class/gpio/unexport", pin.to_s)
end
end
def value
File.read("#{base_path}/value").chomp
end
def value=(v)
File.binwrite "#{base_path}/value", v.to_s
end
def direction
File.read("#{base_path}/direction").chomp
end
def direction=(v)
if direction != v
File.binwrite "#{base_path}/direction", v.to_s
end
end
def edge
File.read("#{base_path}/edge").chomp
end
def edge=(v)
# puts "Not setting edge #{v}"
File.binwrite "#{base_path}/edge", v.to_s
end
def base_path
"/sys/class/gpio/gpio#{pin}"
end
end
Button = Struct.new(:name, :pin) do
def value
gpio.value
end
def value_path
gpio.base_path + "/value"
end
def pressed!
# noop
end
def released!
# noop
end
def setup
puts "Setting up #{self.inspect}..."
system "python setup-button.py #{pin}"
sleep 1
gpio.setup
gpio.direction = "in"
# "none", "rising", "falling", or "both"
gpio.edge = "both"
end
def teardown
gpio.teardown
end
def gpio
@gpio ||= Gpio.new(pin)
end
end
class RoomButton < Button
def pressed!
Speaker.find(name: name).toggle
end
end
class FavoriteButton < Button
def pressed!
Thread.new {
Favorite.find(name: name).play
}
end
end
Favorite = Struct.new(:id, :name) do
def self.all
@all ||= fetch_all
end
def self.find(name:)
all.find { |f| f.name == name }
end
def self.fetch_all
CLIENT.get("/households/#{HOUSEHOLD}/favorites").fetch("items").map { |item| new(item.fetch("id"), item.fetch("name")) }
end
def self.reset!
@all ||= fetch_all
end
def play
CLIENT.post("/groups/#{(Group.playing || Group.first).id}/favorites", favoriteId: id.to_s, playOnCompletion: true)
end
end
Led = Struct.new(:name, :pin) do
ON = "1"
OFF = "0"
def on
gpio.value = ON
end
def off
gpio.value = OFF
end
def on?
value == ON
end
def off?
value == OFF
end
def value
gpio.value
end
def setup
puts "Setting up #{self.inspect}..."
gpio.setup
gpio.direction = "out"
puts "Done."
end
def teardown
gpio.teardown
end
def gpio
@gpio ||= Gpio.new(pin)
end
end
def watch_buttons(buttons)
epoll = Epoll.create
buttons.each do |button|
# Read the initial pin value
fd = File.open(button.value_path, 'r')
puts "Reading first value..."
puts fd.read.chomp
fd.seek 0, IO::SEEK_SET
epoll.add fd, Epoll::PRI
fd
end
loop do
begin
puts "poll wait...."
evlist = epoll.wait # put the program to sleep until the status changes
puts "button pressed!"
evlist.each do |ev|
fd = ev.data
button = BUTTONS.find { |b| b.value_path == fd.path }
if fd.read.chomp == "0"
p button
button.pressed!
else
button.released!
end
fd.seek 0, IO::SEEK_SET
end
rescue
puts $!
Group.reset!
end
end
end
BUTTONS = [
KITCHEN_BUTTON = RoomButton.new("Kitchen", 4),
LIVING_ROOM_BUTTON = RoomButton.new("Living Room", 2),
FIP_BUTTON = FavoriteButton.new("FIP", 25),
REGGAE_BUTTON = FavoriteButton.new("FIP autour du reggae", 24),
ICI_BUTTON = FavoriteButton.new("ICI Musique Vancouver", 23),
DISCOVER_BUTTON = FavoriteButton.new("Discover Weekly", 18),
LILA_BUTTON = FavoriteButton.new("Lila", 15),
# OTHER_BUTTON = FavoriteButton.new("Other", 14)
]
LEDS = [
KITCHEN_LED = Led.new("Kitchen", 17),
LIVING_ROOM_LED = Led.new("Living Room", 3)
]
def sync_leds
loop do
begin
Group.reset!
Speaker.all.each do |speaker|
led = LEDS.find { |led| led.name == speaker.name }
if speaker.playing?
led.on unless led.on?
else
led.off unless led.off?
end
end
sleep 1
rescue
puts $!
sleep 1
end
end
end
def unexport
(BUTTONS + LEDS).each(&:teardown)
end
if ARGV[0] == "console"
require 'irb'
binding.irb
exit
end
require 'epoll'
if ARGV[0] == "board"
puts "Setup GPIO..."
(BUTTONS + LEDS).each(&:setup)
Thread.new do
sync_leds
end
watch_buttons(BUTTONS)
end
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment