1# -*- coding: utf-8 -*- 2# Copyright 2006 Joe Wreschnig 3# 4# This program is free software; you can redistribute it and/or modify 5# it under the terms of the GNU General Public License as published by 6# the Free Software Foundation; either version 2 of the License, or 7# (at your option) any later version. 8 9"""Read and write Ogg Speex comments. 10 11This module handles Speex files wrapped in an Ogg bitstream. The 12first Speex stream found is used. 13 14Read more about Ogg Speex at http://www.speex.org/. This module is 15based on the specification at http://www.speex.org/manual2/node7.html 16and clarifications after personal communication with Jean-Marc, 17http://lists.xiph.org/pipermail/speex-dev/2006-July/004676.html. 18""" 19 20__all__ = ["OggSpeex", "Open", "delete"] 21 22from mutagen import StreamInfo 23from mutagen._vorbis import VCommentDict 24from mutagen.ogg import OggPage, OggFileType, error as OggError 25from mutagen._util import cdata, get_size, loadfile, convert_error 26from mutagen._tags import PaddingInfo 27 28 29class error(OggError): 30 pass 31 32 33class OggSpeexHeaderError(error): 34 pass 35 36 37class OggSpeexInfo(StreamInfo): 38 """OggSpeexInfo() 39 40 Ogg Speex stream information. 41 42 Attributes: 43 length (`float`): file length in seconds, as a float 44 channels (`int`): number of channels 45 bitrate (`int`): nominal bitrate in bits per second. The reference 46 encoder does not set the bitrate; in this case, the bitrate will 47 be 0. 48 """ 49 50 length = 0 51 channels = 0 52 bitrate = 0 53 54 def __init__(self, fileobj): 55 page = OggPage(fileobj) 56 while not page.packets[0].startswith(b"Speex "): 57 page = OggPage(fileobj) 58 if not page.first: 59 raise OggSpeexHeaderError( 60 "page has ID header, but doesn't start a stream") 61 self.sample_rate = cdata.uint_le(page.packets[0][36:40]) 62 self.channels = cdata.uint_le(page.packets[0][48:52]) 63 self.bitrate = max(0, cdata.int_le(page.packets[0][52:56])) 64 self.serial = page.serial 65 66 def _post_tags(self, fileobj): 67 page = OggPage.find_last(fileobj, self.serial, finishing=True) 68 if page is None: 69 raise OggSpeexHeaderError 70 self.length = page.position / float(self.sample_rate) 71 72 def pprint(self): 73 return u"Ogg Speex, %.2f seconds" % self.length 74 75 76class OggSpeexVComment(VCommentDict): 77 """Speex comments embedded in an Ogg bitstream.""" 78 79 def __init__(self, fileobj, info): 80 pages = [] 81 complete = False 82 while not complete: 83 page = OggPage(fileobj) 84 if page.serial == info.serial: 85 pages.append(page) 86 complete = page.complete or (len(page.packets) > 1) 87 data = OggPage.to_packets(pages)[0] 88 super(OggSpeexVComment, self).__init__(data, framing=False) 89 self._padding = len(data) - self._size 90 91 def _inject(self, fileobj, padding_func): 92 """Write tag data into the Speex comment packet/page.""" 93 94 fileobj.seek(0) 95 96 # Find the first header page, with the stream info. 97 # Use it to get the serial number. 98 page = OggPage(fileobj) 99 while not page.packets[0].startswith(b"Speex "): 100 page = OggPage(fileobj) 101 102 # Look for the next page with that serial number, it'll start 103 # the comment packet. 104 serial = page.serial 105 page = OggPage(fileobj) 106 while page.serial != serial: 107 page = OggPage(fileobj) 108 109 # Then find all the pages with the comment packet. 110 old_pages = [page] 111 while not (old_pages[-1].complete or len(old_pages[-1].packets) > 1): 112 page = OggPage(fileobj) 113 if page.serial == old_pages[0].serial: 114 old_pages.append(page) 115 116 packets = OggPage.to_packets(old_pages, strict=False) 117 118 content_size = get_size(fileobj) - len(packets[0]) # approx 119 vcomment_data = self.write(framing=False) 120 padding_left = len(packets[0]) - len(vcomment_data) 121 122 info = PaddingInfo(padding_left, content_size) 123 new_padding = info._get_padding(padding_func) 124 125 # Set the new comment packet. 126 packets[0] = vcomment_data + b"\x00" * new_padding 127 128 new_pages = OggPage._from_packets_try_preserve(packets, old_pages) 129 OggPage.replace(fileobj, old_pages, new_pages) 130 131 132class OggSpeex(OggFileType): 133 """OggSpeex(filething) 134 135 An Ogg Speex file. 136 137 Arguments: 138 filething (filething) 139 140 Attributes: 141 info (`OggSpeexInfo`) 142 tags (`mutagen._vorbis.VCommentDict`) 143 """ 144 145 _Info = OggSpeexInfo 146 _Tags = OggSpeexVComment 147 _Error = OggSpeexHeaderError 148 _mimes = ["audio/x-speex"] 149 150 info = None 151 tags = None 152 153 @staticmethod 154 def score(filename, fileobj, header): 155 return (header.startswith(b"OggS") * (b"Speex " in header)) 156 157 158Open = OggSpeex 159 160 161@convert_error(IOError, error) 162@loadfile(method=False, writable=True) 163def delete(filething): 164 """ delete(filething) 165 166 Arguments: 167 filething (filething) 168 Raises: 169 mutagen.MutagenError 170 171 Remove tags from a file. 172 """ 173 174 t = OggSpeex(filething) 175 filething.fileobj.seek(0) 176 t.delete(filething) 177