mirrors/discord.py
1
0
Fork 0
mirror of https://github.com/Rapptz/discord.py.git synced 2025-07-23 11:26:51 +00:00
Code Issues Projects Releases Wiki Activity

Typehint oggparse.py

Browse Source
This commit is contained in:
apple502j 2021-08-22 18:44:41 +09:00 committed by GitHub
parent d09993d7e7
commit e46d974c8a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 21 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
discord/oggparse.py
View File

@ -22,8 +22,12 @@ FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
DEALINGS IN THE SOFTWARE. DEALINGS IN THE SOFTWARE.
""" """
from __future__ import annotations
import struct import struct
from typing import TYPE_CHECKING, ClassVar, IO, Generator, Tuple, Optional
from .errors import DiscordException from .errors import DiscordException
__all__ = ( __all__ = (
@ -40,22 +44,29 @@ class OggError(DiscordException):
# https://tools.ietf.org/html/rfc7845 # https://tools.ietf.org/html/rfc7845
class OggPage: class OggPage:
_header = struct.Struct('<xBQIIIB') _header: ClassVar[struct.Struct] = struct.Struct('<xBQIIIB')
if TYPE_CHECKING:
flag: int
gran_pos: int
serial: int
pagenum: int
crc: int
segnum: int
def __init__(self, stream): def __init__(self, stream: IO[bytes]) -> None:
try: try:
header = stream.read(struct.calcsize(self._header.format)) header = stream.read(struct.calcsize(self._header.format))
self.flag, self.gran_pos, self.serial, \ self.flag, self.gran_pos, self.serial, \
self.pagenum, self.crc, self.segnum = self._header.unpack(header) self.pagenum, self.crc, self.segnum = self._header.unpack(header)
self.segtable = stream.read(self.segnum) self.segtable: bytes = stream.read(self.segnum)
bodylen = sum(struct.unpack('B'*self.segnum, self.segtable)) bodylen = sum(struct.unpack('B'*self.segnum, self.segtable))
self.data = stream.read(bodylen) self.data: bytes = stream.read(bodylen)
except Exception: except Exception:
raise OggError('bad data stream') from None raise OggError('bad data stream') from None
def iter_packets(self): def iter_packets(self) -> Generator[Tuple[bytes, bool], None, None]:
packetlen = offset = 0 packetlen = offset = 0
partial = True partial = True
@ -74,10 +85,10 @@ class OggPage:
yield self.data[offset:], False yield self.data[offset:], False
class OggStream: class OggStream:
def __init__(self, stream): def __init__(self, stream: IO[bytes]) -> None:
self.stream = stream self.stream: IO[bytes] = stream
def _next_page(self): def _next_page(self) -> Optional[OggPage]:
head = self.stream.read(4) head = self.stream.read(4)
if head == b'OggS': if head == b'OggS':
return OggPage(self.stream) return OggPage(self.stream)
@ -86,13 +97,13 @@ class OggStream:
else: else:
raise OggError('invalid header magic') raise OggError('invalid header magic')
def _iter_pages(self): def _iter_pages(self) -> Generator[OggPage, None, None]:
page = self._next_page() page = self._next_page()
while page: while page:
yield page yield page
page = self._next_page() page = self._next_page()
def iter_packets(self): def iter_packets(self) -> Generator[bytes, None, None]:
partial = b'' partial = b''
for page in self._iter_pages(): for page in self._iter_pages():
for data, complete in page.iter_packets(): for data, complete in page.iter_packets():