Source code for tidalpy.streaming
"""Media streaming: hand the *plaintext* audio bytes to the caller.
:class:`TidalClient.stream` returns a :class:`MediaStream` that wraps a
resolved :class:`~tidalpy.models.StreamInfo` and yields decrypted audio
chunks over the wire — no disk I/O anywhere in the library. This is the
integration path for Django/Celery hosts that store media in S3,
PostgreSQL, or any other backend the library knows nothing about::
with client.stream(url) as media:
for chunk in media.chunks():
s3_upload_part(chunk)
``MediaStream`` is safe to read from one thread at a time; it is not an
unbounded buffer — each call to :meth:`chunks`/ :meth:`read` blocks on
the network until the next chunk arrives.
"""
from __future__ import annotations
import logging
import threading
from typing import Callable, Iterator
import requests
from .errors import CancelledError, DownloadError
from .models import StreamInfo, Track
from .transport import Transport
logger = logging.getLogger("tidalpy.streaming")
DEFAULT_CHUNK_SIZE = 256 * 1024
[docs]
class MediaStream:
"""A resolved track's audio bytes, readable as a plaintext stream.
Attributes
----------
track:
The :class:`~tidalpy.models.Track` this stream belongs to.
info:
The :class:`~tidalpy.models.StreamInfo` (codec, quality, urls).
The object is a context manager and an iterator of bytes chunks.
"""
[docs]
def __init__(
self,
track: Track,
info: StreamInfo,
transport: Transport,
stop_event: threading.Event | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
on_close: Callable[[], None] | None = None,
) -> None:
self.track = track
self.info = info
self._transport = transport
self._stop = stop_event or threading.Event()
self._chunk_size = max(16 * 1024, int(chunk_size))
self._on_close = on_close
self._decryptor = None
if info.encryption_key:
from .decrypt import StreamDecryptor
self._decryptor = StreamDecryptor(info.encryption_key)
self._segment_index = 0
self._iterator: Iterator[bytes] | None = None
self._response: requests.Response | None = None
self._size_cache: int | None = None
# Bytes read from the wire but not yet handed to the caller
# (a read(n) that stopped mid-chunk). Must be served first.
self._leftover = b""
# -- facts ---------------------------------------------------------------
@property
def track_id(self) -> int | str | None:
"""The Tidal track id (shorthand for ``track.id``) — hosts use
it to name the stored object."""
return self.track.id
@property
def codec(self) -> str:
return self.info.codec
@property
def delivered_quality(self) -> str:
return self.info.delivered_quality
@property
def size(self) -> int:
"""Total bytes of the logical file, or ``-1`` when unknown.
Probes each segment once (HEAD, then ranged GET) and caches the
result. A failed probe keeps the value at ``-1`` — it is *not*
an error, callers may just not know the length ahead of time.
"""
if self._size_cache is None:
total = 0
for url in self.info.urls:
size = self._transport.probe_size(url)
if size < 0:
self._size_cache = -1
return self._size_cache
total += size
self._size_cache = total
return self._size_cache
def __repr__(self) -> str:
"""A compact shell/log line — the identity and delivery facts,
without probing for the (network) size."""
return (
f"MediaStream(track={self.track_id!r}, "
f"{self.codec} {self.delivered_quality})"
)
# -- reading ---------------------------------------------------------------
def _check_stop(self) -> None:
if self._stop.is_set():
raise CancelledError("stream cancelled")
def _open_next_segment(self) -> None:
while self._segment_index < len(self.info.urls):
url = self.info.urls[self._segment_index]
self._response = self._transport.fetch(url)
self._segment_index += 1
if self._response.status_code == 200:
self._iterator = self._response.iter_content(
chunk_size=self._chunk_size
)
return
status = self._response.status_code
self._close_segment()
raise DownloadError(f"media segment fetch failed (HTTP {status})")
# No more segments: normal end of stream.
self._iterator = None
def _close_segment(self) -> None:
if self._response is not None:
self._response.close()
self._response = None
self._iterator = None
def _next_chunk(self) -> bytes | None:
while True:
self._check_stop()
if self._leftover:
chunk, self._leftover = self._leftover, b""
return chunk
if self._iterator is None:
self._open_next_segment()
if self._iterator is None:
return None
try:
chunk = next(self._iterator)
except StopIteration:
self._close_segment()
continue
if chunk:
if self._decryptor is not None:
chunk = self._decryptor.decrypt(chunk)
return chunk
[docs]
def chunks(self, size: int | None = None) -> Iterator[bytes]:
"""Yield non-empty plaintext chunks until the stream is exhausted."""
if size is not None:
self._chunk_size = max(16 * 1024, int(size))
while True:
chunk = self._next_chunk()
if chunk is None:
return
yield chunk
def __iter__(self) -> Iterator[bytes]:
return self.chunks()
[docs]
def read(self, n: int = -1) -> bytes:
"""Read up to ``n`` bytes (all remaining when ``n`` is negative)."""
parts: list[bytes] = []
budget = None if n is None or n < 0 else n
while budget is None or len(b"".join(parts)) < budget:
chunk = self._next_chunk()
if chunk is None:
break
if budget is not None:
remaining = budget - len(b"".join(parts))
if len(chunk) > remaining:
self._leftover = chunk[remaining:]
chunk = chunk[:remaining]
parts.append(chunk)
return b"".join(parts)
[docs]
def readinto(self, buf: bytearray) -> int:
"""Fill a writable buffer; returns the number of bytes written."""
offset = 0
while offset < len(buf):
chunk = self._next_chunk()
if chunk is None:
break
take = min(len(buf) - offset, len(chunk))
buf[offset : offset + take] = chunk[:take]
if take < len(chunk):
self._leftover = chunk[take:]
offset += take
return offset
# -- lifecycle ---------------------------------------------------------------
[docs]
def close(self) -> None:
"""Close the open segment and release host resources (idempotent)."""
self._close_segment()
if self._on_close is not None:
self._on_close()
self._on_close = None
def __enter__(self) -> "MediaStream":
return self
def __exit__(self, *exc_info) -> None:
self.close()