home *** CD-ROM | disk | FTP | other *** search
# Source Generated with Decompyle++ # File: in.pyc (Python 2.4) '''" This module is used to convert audio data to local WAV files. These files are cached in the local filesystem and can be used by the recording module. ''' import tempfile import os import gst from types import StringType import urlutil import operations import audio class GetMusic(operations.MeasurableOperation, operations.OperationListener): '''" A Measurable Operation that finishes when the filename is available in the music pool. ''' def __init__(self, pool, music): operations.MeasurableOperation.__init__(self) self._GetMusic__music = music self._GetMusic__pool = pool self._GetMusic__oper = None def getProgress(self): if self._GetMusic__oper is not None: return self._GetMusic__oper.progress return 0.0 progress = property(getProgress) running = property((lambda self: if self._GetMusic__oper: passself._GetMusic__oper.running)) can_stop = property((lambda self: if self._GetMusic__oper: passself._GetMusic__oper.can_stop)) music = property((lambda self: self._GetMusic__music)) pool = property((lambda self: self._GetMusic__pool)) def start(self): if self._GetMusic__pool.is_available(self._GetMusic__music): self._send_finished_event(operations.SUCCESSFUL) self._GetMusic__oper = None else: try: self._GetMusic__oper = self._GetMusic__pool.fetch_music(self._GetMusic__music) self._GetMusic__oper.listeners.append(self) self._GetMusic__oper.start() except Exception: e = None import traceback traceback.print_exc() self._send_finished_event(operations.ERROR, error = e) def on_finished(self, event): self._send_finished_event(event.id, event.error) self._GetMusic__oper = None def stop(self): if not self._GetMusic__oper: raise AssertionError self._GetMusic__oper.stop() class MusicPool: ''' A music pool is basically a place where you put your music to convert it to local WAV files which will then be used to create audio CDs. ''' def __init__(self): raise NotImplementedError def fetch_music(self, music): """ Returns a operations.MeasurableOperation that correspondes the user can use to monitor it's progress. """ raise NotImplementedError def is_available(self, music): ''' Returns if a certain music is on cache. ''' raise NotImplementedError def get_filename(self, music): ''' Returns the filename associated with a certain music in cache. The user must be sure that the file is in cache. ''' raise NotImplementedError def clear(self): """ Clears the pool of it's elements. """ raise NotImplementedError class GstCacheEntry: def __init__(self, filename, is_temp): self.is_temp = is_temp self.filename = filename class GstSourceToWavListener(operations.OperationListener): def __init__(self, parent, filename, music): self._GstSourceToWavListener__filename = filename self._GstSourceToWavListener__music = music self._GstSourceToWavListener__parent = parent def on_finished(self, event): if event.id == operations.SUCCESSFUL: self._GstSourceToWavListener__parent.cache[self._GstSourceToWavListener__parent.unique_music_id(self._GstSourceToWavListener__music)] = GstCacheEntry(self._GstSourceToWavListener__filename, True) else: os.unlink(self._GstSourceToWavListener__filename) class GstMusicPool(MusicPool): def __init__(self): self._GstMusicPool__cache = { } self._GstMusicPool__temp_dir = None cache = property((lambda self: self._GstMusicPool__cache)) def setTemporaryDir(self, temp_dir): if not temp_dir == None and isinstance(temp_dir, StringType): raise AssertionError, 'Directory must be a string. Or None for default.' self._GstMusicPool__temp_dir = temp_dir def getTemporaryDir(self): return self._GstMusicPool__temp_dir temporaryDir = property(getTemporaryDir, setTemporaryDir) def is_available(self, music): return self.cache.has_key(self.unique_music_id(music)) def get_filename(self, music): if not self.is_available(music): raise AssertionError return self.cache[self.unique_music_id(music)].filename def get_source(self, music): raise NotImplementedError def fetch_music(self, music): ''' Can throw a OSError exception in case of the provided temporary dir being invalid. ''' if not not self.is_available(music): raise AssertionError source = self.get_source() (handle, filename) = tempfile.mkstemp(suffix = '.wav', dir = self.temporaryDir) os.close(handle) our_listener = GstSourceToWavListener(self, filename, music) oper = audio.convert_to_wav(source, music, filename) oper.listeners.append(our_listener) return oper def clear(self): for key in self.cache: entry = self.cache[key] if entry.is_temp: os.unlink(entry.filename) continue self._GstMusicPool__cache = { } def __del__(self): self.clear() def unique_music_id(self, music): pass class GvfsMusicPool(GstMusicPool): use_gnomevfs = True def unique_music_id(self, uri): """ Provides a way of uniquely identifying URI's, in case of user sends: file:///foo%20bar file:///foo bar /foo bar Returns always a URL, even if the string was a file path. """ return urlutil.normalize(uri) def is_wav(self, filename): is_pcm = audio.is_wav_pcm(self.get_source(), filename) return operations.syncOperation(is_pcm).id == operations.SUCCESSFUL def is_local(self, filename): return urlutil.is_local(filename) def is_available(self, music): on_cache = GstMusicPool.is_available(self, music) unique_id = self.unique_music_id(music) if not on_cache and self.is_local(music) and self.is_wav(unique_id): filename = urlutil.get_path(unique_id) self.cache[unique_id] = GstCacheEntry(filename, False) on_cache = True return on_cache def get_source(self): if self.use_gnomevfs: return audio.GVFS_SRC else: return audio.FILE_SRC class FetchMusicListPriv(operations.OperationListener): def __init__(self, parent, music_list): self.music_list = music_list self.parent = parent def get_music_from_uri(self, uri): for m in self.music_list: if m['location'] == uri: return m continue def on_finished(self, evt): if isinstance(evt.source, operations.OperationsQueue): self.parent._propagate(evt) return None if not isinstance(evt.source, GetMusic): raise AssertionError if evt.id != operations.SUCCESSFUL: return None uri = evt.source.music pool = evt.source.pool filename = pool.get_filename(uri) m = self.get_music_from_uri(uri) if m: m['cache_location'] = filename elif not False: raise AssertionError, "uri '%s' was not found in music list." % uri def before_operation_starts(self, event, operation): e = operations.Event(self.parent) m = self.get_music_from_uri(operation.music) if not m: raise AssertionError for l in self.parent.listeners: if isinstance(l, FetchMusicListListener): l.before_music_fetched(e, m) continue class FetchMusicListListener(operations.OperationListener): def before_music_fetched(self, event, music): pass class FetchMusicList(operations.MeasurableOperation): """ Fetches a music list which contains the field 'uri' and replaces it by a local filename located inside the pool. When the filename is fetched it updates the 'filename' field inside each music entry of the music list. """ def __init__(self, music_list, pool): operations.MeasurableOperation.__init__(self) self._FetchMusicList__music_list = music_list self._FetchMusicList__queue = operations.OperationsQueue() self._FetchMusicList__pool = pool self._FetchMusicList__listener = FetchMusicListPriv(self, music_list) self._FetchMusicList__queue.listeners.append(self._FetchMusicList__listener) def getProgress(self): return self._FetchMusicList__queue.progress progress = property(getProgress) running = property((lambda self: self._FetchMusicList__queue.running)) def start(self): for m in self._FetchMusicList__music_list: get = GetMusic(self._FetchMusicList__pool, m['location']) get.listeners.append(self._FetchMusicList__listener) self._FetchMusicList__queue.append(get) self._FetchMusicList__queue.start() can_stop = property((lambda self: self._FetchMusicList__queue.can_stop)) def stop(self): self._FetchMusicList__queue.stop() if __name__ == '__main__': import sys import gtk import gobject import gtkutil from os import path def quit(): return False def pp(oper): p = oper.get_progress() print p if p == 1: gtk.main_quit() return True class MyListener(FetchMusicListListener): def __init__(self, prog, oper): FetchMusicListListener.__init__(self) self.music = None self.prog = prog self.oper = oper def before_music_fetched(self, evt, music): print music prog_txt = 'Converting ' prog_txt += path.basename(urlutil.get_path(music['location'])) self.prog.sub_progress_text = prog_txt self.prog.progress_fraction = self.oper.progress if self.music: print 'Fetched', self.music['cache_location'] print 'Fetching', music['location'] self.music = music def on_finished(self, e): if self.music: print 'Fetched', self.music print self.oper.progress gtk.main_quit() def tick(self): self.prog.progress_fraction = self.oper.progress return True prog = gtkutil.HigProgress() prog.primary_text = 'Recording Audio Disc' prog.secondary_text = 'Selected files are to be written to a CD or DVD disc. This operation may take a long time, depending on data size and write speed.' prog.show() pool = GvfsMusicPool() music_list = [ { 'location': path.abspath(sys.argv[1]) }] fetcher = FetchMusicList(music_list, pool) l = MyListener(prog, fetcher) fetcher.listeners.append(l) gobject.timeout_add(200, l.tick) fetcher.start() gtk.main() pool.clear() 
(.txt)