Scrameupeutchi/librespot-python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

importing the compiled pb2 modules with alias

Browse Source
This commit is contained in:
Pawan Paudel
2021-05-26 15:44:04 +05:45
parent 81cbd5a588
commit 3fcf78eb23
14 changed files with 133 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
librespot/audio/PlayableContentFeeder.py
View File

@@ -12,8 +12,8 @@ from librespot.common.Utils import Utils
from librespot.core import Session
from librespot.metadata import PlayableId
from librespot.metadata import TrackId
from librespot.proto import Metadata_pb2
from librespot.proto import StorageResolve_pb2
from librespot.proto import Metadata_pb2 as Metadata
from librespot.proto import StorageResolve_pb2 as StorageResolve
class PlayableContentFeeder:
@@ -26,7 +26,7 @@ class PlayableContentFeeder:
def __init__(self, session: Session):
self.session = session
def pick_alternative_if_necessary(self, track: Metadata_pb2.Track):
def pick_alternative_if_necessary(self, track: Metadata.Track):
if len(track.file) > 0:
return track
@@ -49,7 +49,7 @@ class PlayableContentFeeder:
def resolve_storage_interactive(
self, file_id: bytes,
preload: bool) -> StorageResolve_pb2.StorageResolveResponse:
preload: bool) -> StorageResolve.StorageResolveResponse:
resp = self.session.api().send(
"GET",
(self.STORAGE_RESOLVE_INTERACTIVE_PREFETCH
@@ -65,13 +65,13 @@ class PlayableContentFeeder:
if body is None:
RuntimeError("Response body is empty!")
storage_resolve_response = StorageResolve_pb2.StorageResolveResponse()
storage_resolve_response = StorageResolve.StorageResolveResponse()
storage_resolve_response.ParseFromString(body)
return storage_resolve_response
def load_track(
self,
track_id_or_track: typing.Union[TrackId, Metadata_pb2.Track],
track_id_or_track: typing.Union[TrackId, Metadata.Track],
audio_quality_picker: AudioQualityPicker,
preload: bool,
halt_listener: HaltListener,
@@ -94,9 +94,9 @@ class PlayableContentFeeder:
def load_stream(
self,
file: Metadata_pb2.AudioFile,
track: Metadata_pb2.Track,
episode: Metadata_pb2.Episode,
file: Metadata.AudioFile,
track: Metadata.Track,
episode: Metadata.Episode,
preload: bool,
halt_lister: HaltListener,
):
@@ -104,41 +104,41 @@ class PlayableContentFeeder:
raise RuntimeError()
resp = self.resolve_storage_interactive(file.file_id, preload)
if resp.result == StorageResolve_pb2.StorageResolveResponse.Result.CDN:
if resp.result == StorageResolve.StorageResolveResponse.Result.CDN:
if track is not None:
return CdnFeedHelper.load_track(self.session, track, file,
resp, preload, halt_lister)
return CdnFeedHelper.load_episode(self.session, episode, file,
resp, preload, halt_lister)
elif resp.result == StorageResolve_pb2.StorageResolveResponse.Result.STORAGE:
elif resp.result == StorageResolve.StorageResolveResponse.Result.STORAGE:
if track is None:
# return StorageFeedHelper
pass
elif resp.result == StorageResolve_pb2.StorageResolveResponse.Result.RESTRICTED:
elif resp.result == StorageResolve.StorageResolveResponse.Result.RESTRICTED:
raise RuntimeError("Content is restricted!")
elif resp.result == StorageResolve_pb2.StorageResolveResponse.Response.UNRECOGNIZED:
elif resp.result == StorageResolve.StorageResolveResponse.Response.UNRECOGNIZED:
raise RuntimeError("Content is unrecognized!")
else:
raise RuntimeError("Unknown result: {}".format(resp.result))
class LoadedStream:
episode: Metadata_pb2.Episode
track: Metadata_pb2.Track
episode: Metadata.Episode
track: Metadata.Track
input_stream: GeneralAudioStream
normalization_data: NormalizationData
metrics: PlayableContentFeeder.Metrics
def __init__(
self,
track_or_episode: typing.Union[Metadata_pb2.Track, Metadata_pb2.Episode],
track_or_episode: typing.Union[Metadata.Track, Metadata.Episode],
input_stream: GeneralAudioStream,
normalization_data: NormalizationData,
metrics: PlayableContentFeeder.Metrics,
):
if type(track_or_episode) is Metadata_pb2.Track:
if type(track_or_episode) is Metadata.Track:
self.track = track_or_episode
self.episode = None
elif type(track_or_episode) is Metadata_pb2.Episode:
elif type(track_or_episode) is Metadata.Episode:
self.track = None
self.episode = track_or_episode
else:

6
librespot/audio/StreamId.py
View File

@@ -1,5 +1,5 @@
from librespot.common.Utils import Utils
from librespot.proto import Metadata_pb2
from librespot.proto import Metadata_pb2 as Metadata
class StreamId:
@@ -7,8 +7,8 @@ class StreamId:
episode_gid: bytes = None
def __init__(self,
file: Metadata_pb2.AudioFile = None,
episode: Metadata_pb2.Episode = None):
file: Metadata.AudioFile = None,
episode: Metadata.Episode = None):
if file is None and episode is None:
return
if file is not None:

20
librespot/audio/cdn/CdnFeedHelper.py
View File

@@ -10,23 +10,23 @@ from librespot.audio import NormalizationData
from librespot.audio import PlayableContentFeeder
from librespot.common import Utils
from librespot.core import Session
from librespot.proto import Metadata_pb2
from librespot.proto import StorageResolve_pb2
from librespot.proto import Metadata_pb2 as Metadata
from librespot.proto import StorageResolve_pb2 as StorageResolve
class CdnFeedHelper:
_LOGGER: logging = logging.getLogger(__name__)
@staticmethod
def get_url(resp: StorageResolve_pb2.StorageResolveResponse) -> str:
def get_url(resp: StorageResolve.StorageResolveResponse) -> str:
return random.choice(resp.cdnurl)
@staticmethod
def load_track(
session: Session,
track: Metadata_pb2.Track,
file: Metadata_pb2.AudioFile,
resp_or_url: typing.Union[StorageResolve_pb2.StorageResolveResponse, str],
track: Metadata.Track,
file: Metadata.AudioFile,
resp_or_url: typing.Union[StorageResolve.StorageResolveResponse, str],
preload: bool,
halt_listener: HaltListener,
) -> PlayableContentFeeder.PlayableContentFeeder.LoadedStream:
@@ -54,7 +54,7 @@ class CdnFeedHelper:
@staticmethod
def load_episode_external(
session: Session, episode: Metadata_pb2.Episode,
session: Session, episode: Metadata.Episode,
halt_listener: HaltListener
) -> PlayableContentFeeder.PlayableContentFeeder.LoadedStream:
resp = session.client().head(episode.external_url)
@@ -79,9 +79,9 @@ class CdnFeedHelper:
@staticmethod
def load_episode(
session: Session,
episode: Metadata_pb2.Episode,
file: Metadata_pb2.AudioFile,
resp_or_url: typing.Union[StorageResolve_pb2.StorageResolveResponse, str],
episode: Metadata.Episode,
file: Metadata.AudioFile,
resp_or_url: typing.Union[StorageResolve.StorageResolveResponse, str],
halt_listener: HaltListener,
) -> PlayableContentFeeder.PlayableContentFeeder.LoadedStream:
if type(resp_or_url) is str:

12
librespot/audio/cdn/CdnManager.py
View File

@@ -17,14 +17,14 @@ from librespot.audio.decrypt import NoopAudioDecrypt
from librespot.audio.format import SuperAudioFormat
from librespot.audio.storage import ChannelManager
from librespot.common import Utils
from librespot.proto import StorageResolve_pb2
from librespot.proto import StorageResolve_pb2 as StorageResolve
if typing.TYPE_CHECKING:
from librespot.audio.decrypt.AudioDecrypt import AudioDecrypt
from librespot.audio.HaltListener import HaltListener
from librespot.cache.CacheManager import CacheManager
from librespot.core.Session import Session
from librespot.proto import Metadata_pb2
from librespot.proto import Metadata_pb2 as Metadata
class CdnManager:
@@ -50,7 +50,7 @@ class CdnManager:
return body
def stream_external_episode(self, episode: Metadata_pb2.Episode,
def stream_external_episode(self, episode: Metadata.Episode,
external_url: str,
halt_listener: HaltListener):
return CdnManager.Streamer(
@@ -65,7 +65,7 @@ class CdnManager:
def stream_file(
self,
file: Metadata_pb2.AudioFile,
file: Metadata.AudioFile,
key: bytes,
url: str,
halt_listener: HaltListener,
@@ -96,9 +96,9 @@ class CdnManager:
if body is None:
raise IOError("Response body is empty!")
proto = StorageResolve_pb2.StorageResolveResponse()
proto = StorageResolve.StorageResolveResponse()
proto.ParseFromString(body)
if proto.result == StorageResolve_pb2.StorageResolveResponse.Result.CDN:
if proto.result == StorageResolve.StorageResolveResponse.Result.CDN:
url = random.choice(proto.cdnurl)
self._LOGGER.debug("Fetched CDN url for {}: {}".format(
Utils.bytes_to_hex(file_id), url))

4
librespot/audio/format/AudioQualityPicker.py
View File

@@ -3,10 +3,10 @@ from __future__ import annotations
import typing
if typing.TYPE_CHECKING:
from librespot.proto import Metadata_pb2
from librespot.proto import Metadata_pb2 as Metadata
class AudioQualityPicker:
def get_file(self,
files: typing.List[Metadata_pb2.AudioFile]) -> Metadata_pb2.AudioFile:
files: typing.List[Metadata.AudioFile]) -> Metadata.AudioFile:
pass

26
librespot/audio/format/SuperAudioFormat.py
View File

@@ -1,4 +1,4 @@
from librespot.proto import Metadata_pb2
from librespot.proto import Metadata_pb2 as Metadata
import enum
@@ -8,19 +8,19 @@ class SuperAudioFormat(enum.Enum):
AAC = 0x02
@staticmethod
def get(audio_format: Metadata_pb2.AudioFile.Format):
if audio_format == Metadata_pb2.AudioFile.Format.OGG_VORBIS_96 or \
audio_format == Metadata_pb2.AudioFile.Format.OGG_VORBIS_160 or \
audio_format == Metadata_pb2.AudioFile.Format.OGG_VORBIS_320:
def get(audio_format: Metadata.AudioFile.Format):
if audio_format == Metadata.AudioFile.Format.OGG_VORBIS_96 or \
audio_format == Metadata.AudioFile.Format.OGG_VORBIS_160 or \
audio_format == Metadata.AudioFile.Format.OGG_VORBIS_320:
return SuperAudioFormat.VORBIS
if audio_format == Metadata_pb2.AudioFile.Format.MP3_256 or \
audio_format == Metadata_pb2.AudioFile.Format.MP3_320 or \
audio_format == Metadata_pb2.AudioFile.Format.MP3_160 or \
audio_format == Metadata_pb2.AudioFile.Format.MP3_96 or \
audio_format == Metadata_pb2.AudioFile.Format.MP3_160_ENC:
if audio_format == Metadata.AudioFile.Format.MP3_256 or \
audio_format == Metadata.AudioFile.Format.MP3_320 or \
audio_format == Metadata.AudioFile.Format.MP3_160 or \
audio_format == Metadata.AudioFile.Format.MP3_96 or \
audio_format == Metadata.AudioFile.Format.MP3_160_ENC:
return SuperAudioFormat.MP3
if audio_format == Metadata_pb2.AudioFile.Format.AAC_24 or \
audio_format == Metadata_pb2.AudioFile.Format.AAC_48 or \
audio_format == Metadata_pb2.AudioFile.Format.AAC_24_NORM:
if audio_format == Metadata.AudioFile.Format.AAC_24 or \
audio_format == Metadata.AudioFile.Format.AAC_48 or \
audio_format == Metadata.AudioFile.Format.AAC_24_NORM:
return SuperAudioFormat.AAC
raise RuntimeError("Unknown audio format: {}".format(audio_format))