Automaattiset herätykset iCalista SqueezeBoxiin

Kyllästyin asettelemaan herätyskelloon herätysaikaa manuaalisesti joka päivä, joten hommasin makuuhuoneeseen SqueezeBox Radion ja integroin sen Mac OS X:n iCal-kalenteriin.

Tein tätä varten iCaliin erillisen "Alarms"-kalenterin, johon herätykset asetetaan kalenterimerkintöinä. Niiden URL-kenttään voi lisäksi laittaa herätyksessä soitettavan playlistin. Jos se on tyhjä, oletukseksi tulee DI FM Progressive -nettiradio.

Integrointi osoittautui suhteellisen helpoksi pienellä Python-skriptillä, joka hakee ensin iCalista viikon kalenterimerkinnät ja syöttää ne sitten samalla koneella pyörivään SqueezeBox Serveriin CLI-rajapinnan kautta. Mac OS X tukee Snow Leopardista (10.6) lähtien natiivisti kaikkia tarvittavia Python-rajapintoja. SqueezeBoxin CLI puolestaan on yksinkertainen TCP-yhteys, johon kirjoitetaan ja luetaan tekstirivejä.

#!/usr/bin/python
# vim: sts=4:et
# squeezealarms.py - Generate SqueezeBox alarms from iCal events. (C) 2009 Kenneth Falck <kennu@iki.fi>
import urllib
import socket
import sys
from datetime import datetime, date, timedelta
from CalendarStore import CalCalendarStore, NSPredicate
from Foundation import NSDate, NSCalendar, NSHourCalendarUnit, NSMinuteCalendarUnit, NSSecondCalendarUnit, NSWeekdayCalendarUnit
CALENDAR_NAME = 'Alarms'
ALARM_PLAYLIST_URL = 'http://opml.radiotime.com/Tune.ashx?id=s49631&formats=aac,mp3,wma,wmpro,wmvoice,wmvideo,ogg&partnerId=16'
class SqueezeClient(object):
def init(self, host, port):
self.host = host
self.port = port
self.buffer = ''
self.alarms_list = []
self.alarms_count = 0
self.new_alarms_list = []
self.new_alarms_count = 0
self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.client.connect((self.host, self.port))
self.active = None
def on_alarm(self, player, cmd, id=None, **kwargs):
    if self.active == &#39;create_alarms&#39;:
        self.new_alarms_count += 1
        if self.new_alarms_count &gt;= len(self.new_alarms_list):
            self.active = None
    elif self.active == &#39;delete_all_alarms&#39;:
        self.alarms_count -= 1
        if self.alarms_count &lt;= 0:
            self.active = None

def on_alarms(self, player, start, itemsPerResponse, count, id=None, **kwargs):
    if id is not None:
        self.alarms_count += 1
        self.alarms_list.append(id)
    if self.alarms_count &gt;= int(count):
        # Got whole list.
        if self.active == &#39;delete_all_alarms&#39;:
            if len(self.alarms_list) &lt;= 0:
                self.active = None
            else:
                for alarm_id in self.alarms_list:
                    self.send_command(&#39;alarm delete id:&#39; + alarm_id)
    else:
        if self.active == &#39;delete_all_alarms&#39;:
            self.send_command(&#39;alarms &#39; + str(self.alarms_count) + &#39; 1 filter:all&#39;)

def create_alarms(self, new_alarms_list):
    self.new_alarms_count = 0
    self.new_alarms_list = new_alarms_list
    for alarm in self.new_alarms_list:
        cmd = &#39;alarm add enabled:1 repeat:0 &#39; + &#39; &#39;.join([urllib.quote(key) + &#39;:&#39; + urllib.quote(str(value)) for (key,value) in alarm.items()])
        self.send_command(cmd)
    self.run(&#39;create_alarms&#39;)

def delete_all_alarms(self):
    self.alarms_count = 0
    self.send_command(&#39;alarms 0 1 filter:all&#39;)
    self.run(&#39;delete_all_alarms&#39;)

def send_command(self, cmd):
    self.client.sendall(cmd + &#39;\n&#39;)

def handle_response(self, oargs):
    kwargs = dict([arg.split(&#39;:&#39;, 1) for arg in oargs[1:] if &#39;:&#39; in arg])
    args = [arg for arg in oargs[1:] if &#39;:&#39; not in arg]
    if hasattr(self, &#39;on_&#39; + args[0]):
        getattr(self, &#39;on_&#39; + args[0])(oargs[0], *args[1:], **kwargs)
    else:
        print(&#39;Unhandled command: &#39; + args[0])

def run(self, cmd):
    self.active = cmd
    line = self.client.recv(4096)
    while line is not None and self.active is not None:
        self.buffer += line
        while &#39;\n&#39; in self.buffer:
            (first, self.buffer) = self.buffer.split(&#39;\n&#39;, 1)
            self.handle_response([urllib.unquote(arg) for arg in first.split(&#39; &#39;)])
        if self.active is not None: line = self.client.recv(4096)

def close(self):
    if self.client is not None: self.client.close()
    self.client = None

def __enter__(self):
    return self

def __exit__(self, type, value, traceback):
    self.close()

if name == 'main':
# Read iCal store
now = date.today()
store = CalCalendarStore.defaultCalendarStore()
predicate = NSPredicate.predicateWithFormat_('title == "%s"' % CALENDAR_NAME)
calendars = store.calendars().filteredArrayUsingPredicate_(predicate)
predicate = CalCalendarStore.eventPredicateWithStartDate_endDate_calendars_(now, now+timedelta(days=6), calendars)
events = store.eventsWithPredicate_(predicate)
alarms = []
for event in events:
s = NSCalendar.currentCalendar().components_fromDate_(NSHourCalendarUnit | NSMinuteCalendarUnit | NSSecondCalendarUnit | NSWeekdayCalendarUnit, event.startDate())
url = event.url()
if url is not None: url = str(url)
if url is None or len(url) <= 0: url = ALARM_PLAYLIST_URL
alarms.append(dict(
time = s.hour()*3600 + s.minute()*60 + s.second(),
dow = s.weekday()-1,
url = url
))
# Create alarms on SqueezeBox
with SqueezeClient(&#39;127.0.0.1&#39;, 9090) as client:
    client.delete_all_alarms()
    client.create_alarms(alarms)</code></pre>

Katsotaan heräänkö huomenna ajoissa töihin tämän virityksen jälkeen ;-)