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 == 'create_alarms':
            self.new_alarms_count += 1
            if self.new_alarms_count >= len(self.new_alarms_list):
                self.active = None
        elif self.active == 'delete_all_alarms':
            self.alarms_count -= 1
            if self.alarms_count <= 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 >= int(count):
            # Got whole list.
            if self.active == 'delete_all_alarms':
                if len(self.alarms_list) <= 0:
                    self.active = None
                else:
                    for alarm_id in self.alarms_list:
                        self.send_command('alarm delete id:' + alarm_id)
        else:
            if self.active == 'delete_all_alarms':
                self.send_command('alarms ' + str(self.alarms_count) + ' 1 filter:all')

    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 = 'alarm add enabled:1 repeat:0 ' + ' '.join([urllib.quote(key) + ':' + urllib.quote(str(value)) for (key,value) in alarm.items()])
            self.send_command(cmd)
        self.run('create_alarms')

    def delete_all_alarms(self):
        self.alarms_count = 0
        self.send_command('alarms 0 1 filter:all')
        self.run('delete_all_alarms')

    def send_command(self, cmd):
        self.client.sendall(cmd + '\n')

    def handle_response(self, oargs):
        kwargs = dict([arg.split(':', 1) for arg in oargs[1:] if ':' in arg])
        args = [arg for arg in oargs[1:] if ':' not in arg]
        if hasattr(self, 'on_' + args[0]):
            getattr(self, 'on_' + args[0])(oargs[0], *args[1:], **kwargs)
        else:
            print('Unhandled command: ' + 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 '\n' in self.buffer:
                (first, self.buffer) = self.buffer.split('\n', 1)
                self.handle_response([urllib.unquote(arg) for arg in first.split(' ')])
            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('127.0.0.1', 9090) as client:
        client.delete_all_alarms()
        client.create_alarms(alarms)

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

Published 1.11.2009