source: trunk/StreamVision/StreamVision.py@ 669

Last change on this file since 669 was 669, checked in by Nicholas Riley, 11 years ago

StreamVision.py: Fix OneFileCache.

File size: 14.8 KB
Line 
1#!/usr/bin/pythonw
2# -*- coding: utf-8 -*-
3
4from aem.ae import newdesc
5from appscript import app, k, its, CommandError
6from AppKit import (NSApplication, NSApplicationDefined, NSBeep, NSImage,
7 NSSystemDefined, NSURL, NSWorkspace)
8from Foundation import (NSDistributedNotificationCenter,
9 NSSearchPathForDirectoriesInDomains,
10 NSCachesDirectory, NSUserDomainMask)
11from PyObjCTools import AppHelper
12from Carbon.CarbonEvt import RegisterEventHotKey, GetApplicationEventTarget
13from Carbon.Events import cmdKey, shiftKey, controlKey
14from AudioDevice import (default_output_device_is_airplay,
15 set_default_output_device_changed_callback)
16import httplib2
17import os
18import scrape
19import HotKey
20
21GROWL_APP_NAME = 'StreamVision'
22NOTIFICATION_TRACK_INFO = 'iTunes Track Info'
23NOTIFICATIONS_ALL = [NOTIFICATION_TRACK_INFO]
24
25kEventHotKeyPressedSubtype = 6
26kEventHotKeyReleasedSubtype = 9
27
28kHIDUsage_Csmr_ScanNextTrack = 0xB5
29kHIDUsage_Csmr_ScanPreviousTrack = 0xB6
30kHIDUsage_Csmr_PlayOrPause = 0xCD
31
32def growlRegister():
33 global growl
34 growl = app(id='com.Growl.GrowlHelperApp')
35
36 growl.register(
37 as_application=GROWL_APP_NAME,
38 all_notifications=NOTIFICATIONS_ALL,
39 default_notifications=NOTIFICATIONS_ALL,
40 icon_of_application='iTunes.app')
41 # if we leave off the .app, we can get Classic iTunes's icon
42
43def growlNotify(title, description, **kw):
44 try:
45 if usingStereo:
46 description += '\n(AirPlay)'
47
48 if 'image' in kw:
49 image = (NSImage.alloc().initWithData_(buffer(kw['image']))
50 .TIFFRepresentation())
51 kw['image'] = newdesc('TIFF', image)
52
53 growl.notify(
54 with_name=NOTIFICATION_TRACK_INFO,
55 title=title,
56 description=description,
57 application_name=GROWL_APP_NAME,
58 **kw)
59 except CommandError:
60 growlRegister()
61 growlNotify(title, description, **kw)
62
63def radioParadiseURL():
64 session = scrape.Session()
65 session.go('http://radioparadise.com/jq_playlist.php')
66 url = session.region.firsttag('a')['href']
67 if not url.startswith('http'):
68 url = 'http://www.radioparadise.com/rp2-' + url
69 return url
70
71def cleanStreamTitle(title):
72 if title == k.missing_value:
73 return ''
74 title = title.split(' [')[0] # XXX move to description
75 try: # incorrectly encoded?
76 title = title.encode('iso-8859-1').decode('utf-8')
77 except (UnicodeDecodeError, UnicodeEncodeError):
78 pass
79 title = title.replace('`', u'’')
80 return title
81
82def cleanStreamTrackName(name):
83 name = name.split('. ')[0]
84 name = name.split(': ')[0]
85 name = name.split(' - ')
86 if len(name) > 1:
87 name = ' - '.join(name[:-1])
88 else:
89 name = name[0]
90 return name
91
92def iTunesApp(): return app(id='com.apple.iTunes')
93def XTensionApp(): return app(creator='SHEx')
94def AmuaApp(): return app('Amua.app')
95
96HAVE_XTENSION = False
97try:
98 XTensionApp()
99 HAVE_XTENSION = True
100except:
101 pass
102
103HAVE_AMUA = False
104try:
105 AmuaApp()
106 HAVE_AMUA = True
107except:
108 pass
109
110needsStereoPowerOn = HAVE_XTENSION
111usingStereo = False
112
113def mayUseStereo():
114 if not HAVE_XTENSION:
115 return False
116 try:
117 # A bit better in iTunes 11.0.3, but can't do this via an Apple
118 # Event descriptor; have to send multiple events
119 return any(d.kind() != k.computer
120 for d in iTunesApp().current_AirPlay_devices())
121 except AttributeError:
122 pass
123 except CommandError: # iTunes 11.1 fails with AirPort Express
124 return True
125
126 systemEvents = app(id='com.apple.systemEvents')
127 iTunesWindow = systemEvents.application_processes[u'iTunes'].windows[u'iTunes']
128 # Can't get AirPlay status with iTunes Mini Player or window on other Space.
129 try:
130 remote_speakers = iTunesWindow.buttons[its.attributes['AXDescription'].value.beginswith(u'AirPlay')].title()
131 except CommandError: # window on another Space?
132 return usingStereo
133 return remote_speakers and remote_speakers[0] not in (None, k.missing_value)
134
135def turnStereoOnOrOff():
136 global needsStereoPowerOn, usingStereo
137 usingStereo = False
138 if not default_output_device_is_airplay() and not mayUseStereo():
139 if HAVE_XTENSION and XTensionApp().status('Stereo'):
140 XTensionApp().turnoff('Stereo')
141 return
142 if not XTensionApp().status('Stereo'):
143 XTensionApp().turnon('Stereo')
144 usingStereo = True
145 needsStereoPowerOn = False
146
147def turnStereoOff():
148 global needsStereoPowerOn, usingStereo
149 usingStereo = False
150 if default_output_device_is_airplay() or not mayUseStereo():
151 return
152 if not needsStereoPowerOn and XTensionApp().status('Stereo'):
153 XTensionApp().turnoff('Stereo')
154 needsStereoPowerOn = True
155
156def amuaPlaying():
157 if not HAVE_AMUA:
158 return False
159 return AmuaApp().is_playing()
160
161class OneFileCache(object):
162 __slots__ = ('key', 'cache')
163
164 def __init__(self, cache):
165 if not os.path.exists(cache):
166 os.makedirs(cache)
167 self.cache = os.path.join(cache, 'file')
168 self.key = None
169
170 def get(self, key):
171 if key == self.key:
172 return file(self.cache, 'r').read()
173
174 def set(self, key, value):
175 self.key = key
176 file(self.cache, 'w').write(value)
177
178 def delete(self, key):
179 if key == self.key:
180 self.key = None
181 os.remove(self.cache)
182
183class StreamVision(NSApplication):
184
185 hotKeyActions = {}
186 hotKeys = []
187
188 def displayTrackInfo(self, playerInfo=None):
189 iTunes = iTunesApp()
190
191 try:
192 trackClass = iTunes.current_track.class_()
193 except CommandError:
194 trackClass = k.property
195
196 trackName = ''
197 if trackClass != k.property:
198 trackName = iTunes.current_track.name()
199
200 try:
201 playerState = iTunes.player_state()
202 except CommandError:
203 playerState = None # probably iTunes quit
204 if playerState != k.playing:
205 if playerState != None:
206 growlNotify('iTunes is not playing.', trackName)
207 turnStereoOff()
208 return
209 turnStereoOnOrOff()
210 if trackClass == k.URL_track:
211 if amuaPlaying():
212 if playerInfo is None: # Amua displays it itself
213 AmuaApp().display_song_information()
214 return
215 url = iTunes.current_stream_URL()
216 kw = {}
217 if url != k.missing_value and url.endswith('.jpg'):
218 try:
219 response, content = self.http.request(url)
220 except Exception, e:
221 import sys
222 print >> sys.stderr, 'Request for album art failed:', e
223 else:
224 if response['content-type'].startswith('image/'):
225 kw['image'] = content
226 growlNotify(cleanStreamTitle(iTunes.current_stream_title()),
227 cleanStreamTrackName(trackName), **kw)
228 return
229 if trackClass == k.property:
230 growlNotify('iTunes is playing.', '')
231 return
232 kw = {}
233 # XXX iTunes doesn't let you get artwork for shared tracks
234 if trackClass != k.shared_track:
235 artwork = iTunes.current_track.artworks()
236 if artwork:
237 try:
238 kw['image'] = artwork[0].data_().data
239 except CommandError:
240 pass
241 growlNotify(trackName + ' ' +
242 u'★' * (iTunes.current_track.rating() / 20),
243 iTunes.current_track.album() + '\n' +
244 iTunes.current_track.artist(),
245 **kw)
246
247 def defaultOutputDeviceChanged(self):
248 turnStereoOnOrOff()
249 self.displayTrackInfo()
250
251 def goToSite(self):
252 iTunes = iTunesApp()
253 if iTunes.player_state() == k.playing:
254 if amuaPlaying():
255 AmuaApp().display_album_details()
256 return
257 url = iTunes.current_stream_URL()
258 if url != k.missing_value:
259 if 'radioparadise.com' in url and 'review' not in url:
260 url = radioParadiseURL()
261 NSWorkspace.sharedWorkspace().openURL_(NSURL.URLWithString_(url))
262 return
263 NSBeep()
264
265 def registerHotKey(self, func, keyCode, mods=0):
266 hotKeyRef = RegisterEventHotKey(keyCode, mods, (0, 0),
267 GetApplicationEventTarget(), 0)
268 self.hotKeys.append(hotKeyRef)
269 self.hotKeyActions[HotKey.HotKeyAddress(hotKeyRef)] = func
270 return hotKeyRef
271
272 def unregisterHotKey(self, hotKeyRef):
273 self.hotKeys.remove(hotKeyRef)
274 del self.hotKeyActions[HotKey.HotKeyAddress(hotKeyRef)]
275 hotKeyRef.UnregisterEventHotKey()
276
277 def incrementRatingBy(self, increment):
278 iTunes = iTunesApp()
279 if amuaPlaying():
280 if increment < 0:
281 AmuaApp().ban_song()
282 growlNotify('Banned song.', '', icon_of_application='Amua.app')
283 else:
284 AmuaApp().love_song()
285 growlNotify('Loved song.', '', icon_of_application='Amua.app')
286 return
287 rating = iTunes.current_track.rating()
288 rating += increment
289 if rating < 0:
290 rating = 0
291 NSBeep()
292 elif rating > 100:
293 rating = 100
294 NSBeep()
295 iTunes.current_track.rating.set(rating)
296
297 def playPause(self, useStereo=True):
298 global needsStereoPowerOn
299
300 iTunes = iTunesApp()
301 was_playing = (iTunes.player_state() == k.playing)
302 if not useStereo:
303 needsStereoPowerOn = False
304 if was_playing and amuaPlaying():
305 AmuaApp().stop()
306 else:
307 iTunes.playpause()
308 if not was_playing and iTunes.player_state() == k.stopped:
309 # most likely, we're focused on the iPod, so playing does nothing
310 iTunes.browser_windows[1].view.set(iTunes.user_playlists[its.name=='Stations'][1]())
311 iTunes.play()
312 if not useStereo:
313 return
314 if iTunes.player_state() == k.playing:
315 turnStereoOnOrOff()
316 else:
317 turnStereoOff()
318
319 def playPauseFront(self):
320 systemEvents = app(id='com.apple.systemEvents')
321 frontName = systemEvents.processes[its.frontmost == True][1].name()
322 if frontName == 'RealPlayer':
323 realPlayer = app(id='com.RealNetworks.RealPlayer')
324 if len(realPlayer.players()) > 0:
325 if realPlayer.players[1].state() == k.playing:
326 realPlayer.pause()
327 else:
328 realPlayer.play()
329 return
330 elif frontName == 'VLC':
331 app(id='org.videolan.vlc').play() # equivalent to playpause
332 else:
333 self.playPause(useStereo=False)
334
335 def nextTrack(self):
336 if amuaPlaying():
337 AmuaApp().skip_song()
338 return
339 iTunesApp().next_track()
340
341 def registerZoomWindowHotKey(self):
342 self.zoomWindowHotKey = self.registerHotKey(self.zoomWindow, 42, cmdKey | controlKey) # cmd-ctrl-\
343
344 def unregisterZoomWindowHotKey(self):
345 self.unregisterHotKey(self.zoomWindowHotKey)
346 self.zoomWindowHotKey = None
347
348 def zoomWindow(self):
349 # XXX detect if "enable access for assistive devices" needs to be enabled
350 systemEvents = app(id='com.apple.systemEvents')
351 frontName = systemEvents.processes[its.frontmost == True][1].name()
352 if frontName == 'iTunes':
353 systemEvents.processes['iTunes'].menu_bars[1]. \
354 menu_bar_items['Window'].menus.menu_items['Zoom'].click()
355 return
356 elif frontName in ('X11', 'XQuartz', 'Emacs'): # preserve C-M-\
357 self.unregisterZoomWindowHotKey()
358 systemEvents.key_code(42, using=[k.command_down, k.control_down])
359 self.registerZoomWindowHotKey()
360 return
361 frontPID = systemEvents.processes[its.frontmost == True][1].unix_id()
362 try:
363 zoomed = app(pid=frontPID).windows[1].zoomed
364 zoomed.set(not zoomed())
365 except (CommandError, RuntimeError):
366 systemEvents.processes[frontName].windows \
367 [its.subrole == 'AXStandardWindow'].windows[1]. \
368 buttons[its.subrole == 'AXZoomButton'].buttons[1].click()
369
370 def finishLaunching(self):
371 super(StreamVision, self).finishLaunching()
372
373 caches = NSSearchPathForDirectoriesInDomains(NSCachesDirectory,
374 NSUserDomainMask, True)[0]
375 cache = os.path.join(caches, 'StreamVision')
376 self.http = httplib2.Http(OneFileCache(cache), 5)
377 self.imagePath = os.path.join(cache, 'image')
378
379 self.registerHotKey(self.displayTrackInfo, 100) # F8
380 self.registerHotKey(self.goToSite, 100, cmdKey) # cmd-F8
381 self.registerHotKey(self.playPause, 101) # F9
382 self.registerHotKey(lambda: iTunesApp().previous_track(), 109) # F10
383 self.registerHotKey(self.nextTrack, 103) # F11
384 self.registerHotKey(lambda: self.incrementRatingBy(-20), 109, shiftKey) # shift-F10
385 self.registerHotKey(lambda: self.incrementRatingBy(20), 103, shiftKey) # shift-F11
386 self.registerZoomWindowHotKey()
387
388 distributedNotificationCenter = NSDistributedNotificationCenter.defaultCenter()
389 distributedNotificationCenter.addObserver_selector_name_object_(self, self.displayTrackInfo, 'com.apple.iTunes.playerInfo', None)
390 distributedNotificationCenter.addObserver_selector_name_object_(self, self.terminate_, 'com.apple.logoutContinued', None)
391 try:
392 import HIDRemote
393 HIDRemote.connect()
394 except ImportError:
395 print "failed to import HIDRemote (XXX fix - on Intel)"
396 except OSError, e:
397 print "failed to connect to remote: ", e
398
399 set_default_output_device_changed_callback(
400 self.defaultOutputDeviceChanged)
401 turnStereoOnOrOff()
402
403 def sendEvent_(self, theEvent):
404 eventType = theEvent.type()
405 if eventType == NSSystemDefined and \
406 theEvent.subtype() == kEventHotKeyPressedSubtype:
407 self.hotKeyActions[theEvent.data1()]()
408 elif eventType == NSApplicationDefined:
409 key = theEvent.data1()
410 if key == kHIDUsage_Csmr_ScanNextTrack:
411 self.nextTrack()
412 elif key == kHIDUsage_Csmr_ScanPreviousTrack:
413 iTunesApp().previous_track()
414 elif key == kHIDUsage_Csmr_PlayOrPause:
415 self.playPauseFront()
416 super(StreamVision, self).sendEvent_(theEvent)
417
418if __name__ == "__main__":
419 growlRegister()
420 AppHelper.runEventLoop()
421 try:
422 HIDRemote.disconnect()
423 except:
424 pass
Note: See TracBrowser for help on using the repository browser.