from __future__ import annotations
import ctypes
import io
from dataclasses import dataclass
from typing import TYPE_CHECKING, BinaryIO, List, Optional, Union
from pyglet.media.exceptions import CannotSeekException, MediaException
if TYPE_CHECKING:
from pyglet.image import AbstractImage
from pyglet.image.animation import Animation
from pyglet.media.codecs import MediaEncoder
from pyglet.media.drivers.base import MediaEvent
from pyglet.media.player import Player
class AudioFormat:
"""Audio details.
An instance of this class is provided by sources with audio tracks. You
should not modify the fields, as they are used internally to describe the
format of data provided by the source.
Args:
channels (int): The number of channels: 1 for mono or 2 for stereo
(pyglet does not yet support surround-sound sources).
sample_size (int): Bits per sample; only 8 or 16 are supported.
sample_rate (int): Samples per second (in Hertz).
"""
def __init__(self, channels: int, sample_size: int, sample_rate: int) -> None:
self.channels = channels
self.sample_size = sample_size
self.sample_rate = sample_rate
# Convenience
self.bytes_per_frame = (sample_size // 8) * channels
self.bytes_per_second = self.bytes_per_frame * sample_rate
self.bytes_per_sample = self.bytes_per_frame
"""This attribute is kept for compatibility and should not be used due
to a terminology error.
This value contains the bytes per audio frame, and using
`bytes_per_frame` should be preferred.
For the actual amount of bytes per sample, divide `sample_size` by
eight.
"""
def align(self, num_bytes: int) -> int:
"""Align a given amount of bytes to the audio frame size of this
audio format, downwards.
"""
return num_bytes - (num_bytes % self.bytes_per_frame)
def align_ceil(self, num_bytes: int) -> int:
"""Align a given amount of bytes to the audio frame size of this
audio format, upwards.
"""
return num_bytes + (-num_bytes % self.bytes_per_frame)
def timestamp_to_bytes_aligned(self, timestamp: float) -> int:
"""Given a timestamp, return the amount of bytes that an emitter with
this audio format would have to have played to reach it, aligned
to the audio frame size.
"""
return self.align(int(timestamp * self.bytes_per_second))
def __eq__(self, other: object) -> bool:
if isinstance(other, AudioFormat):
return (self.channels == other.channels and
self.sample_size == other.sample_size and
self.sample_rate == other.sample_rate)
return NotImplemented
def __repr__(self) -> str:
return '%s(channels=%d, sample_size=%d, sample_rate=%d)' % (
self.__class__.__name__, self.channels, self.sample_size,
self.sample_rate)
@dataclass
class VideoFormat:
"""Video details.
An instance of this class is provided by sources with a video stream. You
should not modify the fields.
Note that the sample aspect has no relation to the aspect ratio of the
video image. For example, a video image of 640x480 with sample aspect 2.0
should be displayed at 1280x480. It is the responsibility of the
application to perform this scaling.
Args:
width (int): Width of video image, in pixels.
height (int): Height of video image, in pixels.
sample_aspect (float): Aspect ratio (width over height) of a single
video pixel.
frame_rate (float): Frame rate (frames per second) of the video.
.. versionadded:: 1.2
"""
width: int
height: int
sample_aspect: float = 0.0
frame_rate: float | None = None
class AudioData:
"""A single packet of audio data.
This class is used internally by pyglet.
Args:
data (bytes, ctypes array, or supporting buffer protocol): Sample data.
length (int): Size of sample data, in bytes.
timestamp (float): Time of the first sample, in seconds.
duration (float): Total data duration, in seconds.
events (List[:class:`pyglet.media.drivers.base.MediaEvent`]): List of events
contained within this packet. Events are timestamped relative to
this audio packet.
.. deprecated:: 2.0.10
`timestamp` and `duration` are unused and will be removed eventually.
"""
__slots__ = 'data', 'duration', 'events', 'length', 'pointer', 'timestamp'
def __init__(self,
data: bytes | ctypes.Array,
length: int,
timestamp: float = 0.0,
duration: float = 0.0,
events: list[MediaEvent] | None = None) -> None:
if isinstance(data, bytes):
# bytes are treated specially by ctypes and can be cast to a void pointer, get
# their content's address like this
self.pointer = ctypes.cast(data, ctypes.c_void_p).value
elif isinstance(data, ctypes.Array):
self.pointer = ctypes.addressof(data)
else:
try:
self.pointer = ctypes.addressof(ctypes.c_int.from_buffer(data))
except TypeError:
raise TypeError("Unsupported AudioData type.")
self.data = data
# In any case, `data` will support the buffer protocol by delivering at least
# a readable buffer.
self.length = length
self.timestamp = timestamp
self.duration = duration
self.events = [] if events is None else events
@dataclass
class SourceInfo:
"""Source metadata information.
Fields are the empty string or zero if the information is not available.
Args:
title (str): Title
author (str): Author
copyright (str): Copyright statement
comment (str): Comment
album (str): Album name
year (int): Year
track (int): Track number
genre (str): Genre
.. versionadded:: 1.2
"""
title: str = ''
author: str = ''
copyright: str = ''
comment: str = ''
album: str = ''
year: int = 0
track: int = 0
genre: str = ''
class Source:
"""An audio and/or video source.
Args:
audio_format (:class:`.AudioFormat`): Format of the audio in this
source, or ``None`` if the source is silent.
video_format (:class:`.VideoFormat`): Format of the video in this
source, or ``None`` if there is no video.
info (:class:`.SourceInfo`): Source metadata such as title, artist,
etc; or ``None`` if the` information is not available.
.. versionadded:: 1.2
Attributes:
is_player_source (bool): Determine if this source is a player
current source.
Check on a :py:class:`~pyglet.media.player.Player` if this source
is the current source.
"""
_duration = None
_players: List['Player'] = [] # Players created through Source.play
audio_format = None
video_format = None
info = None
is_player_source = False
@property
def duration(self) -> float:
"""float: The length of the source, in seconds.
Not all source durations can be determined; in this case the value
is ``None``.
Read-only.
"""
return self._duration
def play(self) -> 'Player':
"""Play the source.
This is a convenience method which creates a Player for
this source and plays it immediately.
Returns:
:class:`.Player`
"""
from pyglet.media.player import Player # XXX Nasty circular dependency
player = Player()
player.queue(self)
player.play()
Source._players.append(player)
def _on_player_eos():
Source._players.remove(player)
# There is a closure on player. To break up that reference, delete this function.
player.on_player_eos = None
player.delete()
player.on_player_eos = _on_player_eos
return player
def get_animation(self) -> Animation:
"""Import all video frames into memory.
An empty animation will be returned if the source has no video.
Otherwise, the animation will contain all unplayed video frames (the
entire source, if it has not been queued on a player). After creating
the animation, the source will be at EOS (end of stream).
This method is unsuitable for videos running longer than a
few seconds.
.. versionadded:: 1.1
Returns:
:class:`pyglet.image.Animation`
"""
from pyglet.image import Animation, AnimationFrame
if not self.video_format:
# XXX: This causes an assertion in the constructor of Animation
return Animation([])
else:
frames = []
last_ts = 0
next_ts = self.get_next_video_timestamp()
while next_ts is not None:
image = self.get_next_video_frame()
if image is not None:
delay = next_ts - last_ts
frames.append(AnimationFrame(image, delay))
last_ts = next_ts
next_ts = self.get_next_video_timestamp()
return Animation(frames)
def get_next_video_timestamp(self) -> Optional[float]:
"""Get the timestamp of the next video frame.
.. versionadded:: 1.1
Returns:
float: The next timestamp, or ``None`` if there are no more video
frames.
"""
pass
def get_next_video_frame(self) -> Optional['AbstractImage']:
"""Get the next video frame.
.. versionadded:: 1.1
Returns:
:class:`pyglet.image.AbstractImage`: The next video frame image,
or ``None`` if the video frame could not be decoded or there are
no more video frames.
"""
pass
def save(self,
filename: str,
file: Optional[BinaryIO] = None,
encoder: Optional['MediaEncoder'] = None) -> None:
"""Save this Source to a file.
:Parameters:
`filename` : str
Used to set the file format, and to open the output file
if `file` is unspecified.
`file` : file-like object or None
File to write audio data to.
`encoder` : MediaEncoder or None
If unspecified, all encoders matching the filename extension
are tried. If all fail, the exception from the first one
attempted is raised.
"""
if encoder:
return encoder.encode(self, filename, file)
else:
import pyglet.media.codecs
return pyglet.media.codecs.registry.encode(self, filename, file)
# Internal methods that Player calls on the source:
def is_precise(self) -> bool:
"""bool: Whether this source is considered precise.
``x`` bytes on source ``s`` are considered aligned if
``x % s.audio_format.bytes_per_frame == 0``, so there'd be no partial
audio frame in the returned data.
A source is precise if - for an aligned request of ``x`` bytes - it
returns:\\
- If ``x`` or more bytes are available, ``x`` bytes.
- If not enough bytes are available anymore, ``r`` bytes where
``r < x`` and ``r`` is aligned.
A source is **not** precise if it does any of these:
- Return less than ``x`` bytes for an aligned request of ``x``
bytes although data still remains so that an additional request
would return additional :class:`.AudioData` / not ``None``.
- Return more bytes than requested.
- Return an unaligned amount of bytes for an aligned request.
pyglet's internals are guaranteed to never make unaligned
requests, or requests of less than 1024 bytes.
If this method returns ``False``, pyglet will wrap the source in an
alignment-forcing buffer creating additional overhead.
If this method is overridden to return ``True`` although the source
does not comply with the requirements above, audio playback may be
negatively impacted at best and memory access violations occur at
worst.
:Returns:
bool: Whether the source is precise.
"""
return False
def seek(self, timestamp: float) -> None:
"""Seek to given timestamp.
Args:
timestamp (float): Time where to seek in the source. The
``timestamp`` will be clamped to the duration of the source.
"""
raise CannotSeekException()
def get_queue_source(self) -> 'Source':
"""Return the ``Source`` to be used as the queue source for a player.
Default implementation returns ``self``.
Returns:
:class:`Source`
"""
return self
def get_audio_data(self, num_bytes: int, compensation_time=0.0) -> Optional[AudioData]:
"""Get next packet of audio data.
Args:
num_bytes (int): A size hint for the amount of bytes to return,
but the returned amount may be lower or higher.
compensation_time (float): Time in sec to compensate due to a
difference between the master clock and the audio clock.
.. deprecated:: 2.0.10
compensation_time: Will always be given as ``0.0``.
Returns:
:class:`.AudioData`: Next packet of audio data, or ``None`` if
there is no (more) data.
"""
return None
[docs]
class StreamingSource(Source):
"""A source that is decoded as it is being played.
The source can only be played once at a time on any
:class:`~pyglet.media.player.Player`.
"""
[docs]
def get_queue_source(self) -> 'StreamingSource':
"""Return the ``Source`` to be used as the source for a player.
Default implementation returns self.
Returns:
:class:`.Source`
"""
if self.is_player_source:
raise MediaException('This source is already queued on a player.')
self.is_player_source = True
return super().get_queue_source()
[docs]
def delete(self) -> None:
"""Release the resources held by this StreamingSource."""
pass
[docs]
class StaticSource(Source):
"""A source that has been completely decoded in memory.
This source can be queued onto multiple players any number of times.
Construct a :py:class:`~pyglet.media.StaticSource` for the data in
``source``.
Args:
source (Source): The source to read and decode audio and video data
from.
"""
[docs]
def __init__(self, source: Source) -> None:
source = source.get_queue_source()
if source.video_format:
raise NotImplementedError('Static sources not supported for video.')
self.audio_format = source.audio_format
if self.audio_format is None:
self._data = None
self._duration = 0.0
return
# Arbitrary: number of bytes to request at a time.
buffer_size = 1 << 20 # 1 MB
# Naive implementation. Driver-specific implementations may override
# to load static audio data into device (or at least driver) memory.
data = io.BytesIO()
while True:
audio_data = source.get_audio_data(buffer_size)
if audio_data is None:
break
data.write(audio_data.data)
self._data = data.getvalue()
self._duration = len(self._data) / self.audio_format.bytes_per_second
[docs]
def get_queue_source(self) -> Optional['StaticMemorySource']:
if self._data is not None:
return StaticMemorySource(self._data, self.audio_format)
return None
[docs]
def get_audio_data(self, num_bytes: float, compensation_time: float = 0.0) -> Optional[AudioData]:
"""The StaticSource does not provide audio data.
When the StaticSource is queued on a
:class:`~pyglet.media.player.Player`, it creates a
:class:`.StaticMemorySource` containing its internal audio data and
audio format.
Raises:
RuntimeError
"""
raise RuntimeError('StaticSource cannot be queued.')
class StaticMemorySource(StaticSource):
"""
Helper class for default implementation of :class:`.StaticSource`.
Do not use directly. This class is used internally by pyglet.
Args:
data (readable buffer): The audio data.
audio_format (AudioFormat): The audio format.
"""
def __init__(self, data, audio_format: AudioFormat) -> None:
"""Construct a memory source over the given data buffer."""
self._file = io.BytesIO(data)
self._max_offset = len(data)
self.audio_format = audio_format
self._duration = len(data) / float(audio_format.bytes_per_second)
def is_precise(self) -> bool:
return True
def seek(self, timestamp: float) -> None:
"""Seek to given timestamp.
Args:
timestamp (float): Time where to seek in the source.
"""
offset = int(timestamp * self.audio_format.bytes_per_second)
# Align to audio frame to not corrupt audio data.
self._file.seek(self.audio_format.align(offset))
def get_audio_data(self, num_bytes: float, compensation_time: float = 0.0) -> Optional[AudioData]:
"""Get next packet of audio data.
Args:
num_bytes (int): Maximum number of bytes of data to return.
Returns:
:class:`.AudioData`: Next packet of audio data, or ``None`` if
there is no (more) data.
"""
offset_before = self._file.tell()
data = self._file.read(num_bytes)
if not data:
return None
timestamp = float(offset_before) / self.audio_format.bytes_per_second
duration = len(data) / self.audio_format.bytes_per_second
return AudioData(data, len(data), timestamp, duration)
class SourceGroup:
"""Group of like sources to allow gapless playback.
Seamlessly read data from a group of sources to allow for
gapless playback. All sources must share the same audio format.
The first source added sets the format.
"""
def __init__(self) -> None:
self.audio_format = None
self.video_format = None
self.info = None
self.duration = 0.0
self._timestamp_offset = 0.0
self._dequeued_durations = []
self._sources = []
self.is_player_source = False
def is_precise(self) -> bool:
return False
def seek(self, time: float) -> None:
if self._sources:
self._sources[0].seek(time)
def add(self, source: Source) -> None:
self.audio_format = self.audio_format or source.audio_format
self.info = self.info or source.info
source = source.get_queue_source()
assert (source.audio_format == self.audio_format), "Sources must share the same audio format."
self._sources.append(source)
self.duration += source.duration
def has_next(self) -> bool:
return len(self._sources) > 1
def get_queue_source(self) -> 'SourceGroup':
return self
def _advance(self) -> None:
if self._sources:
self._timestamp_offset += self._sources[0].duration
self._dequeued_durations.insert(0, self._sources[0].duration)
old_source = self._sources.pop(0)
self.duration -= old_source.duration
if isinstance(old_source, StreamingSource):
old_source.delete()
def get_audio_data(self, num_bytes: float, compensation_time=0.0) -> Optional[AudioData]:
"""Get next audio packet.
:Parameters:
`num_bytes` : int
Hint for preferred size of audio packet; may be ignored.
:rtype: `AudioData`
:return: Audio data, or None if there is no more data.
"""
if not self._sources:
return None
buffer = b""
duration = 0.0
timestamp = 0.0
while len(buffer) < num_bytes and self._sources:
audiodata = self._sources[0].get_audio_data(num_bytes)
if audiodata:
buffer += audiodata.data
duration += audiodata.duration
timestamp += self._timestamp_offset
else:
self._advance()
return AudioData(buffer, len(buffer), timestamp, duration)
