Jonathan
68 lines
1.8 KiB
Python
68 lines
1.8 KiB
Python
from typing import Callable
|
|
|
|
from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality
|
|
|
|
_QUALITIES = {
|
|
"qobuz": Quality(fmt="FLAC", lossless=True, bit_depth=24, sample_rate=96000),
|
|
"soulseek": Quality(fmt="FLAC", lossless=True, bit_depth=16, sample_rate=44100),
|
|
"youtube": Quality(fmt="OPUS", lossless=False, bitrate_kbps=160),
|
|
}
|
|
|
|
|
|
class _BaseFake:
|
|
name = "fake"
|
|
tier = 99
|
|
|
|
def __init__(self, matches: bool = True):
|
|
self._matches = matches
|
|
|
|
def health(self) -> bool:
|
|
return True
|
|
|
|
def search(self, target: MBTarget) -> list[Candidate]:
|
|
if not self._matches:
|
|
return []
|
|
tracks = target.track_count or 10
|
|
return [
|
|
Candidate(
|
|
source=self.name,
|
|
source_ref=f"{self.name}:{target.artist}:{target.album}",
|
|
matched_artist=target.artist,
|
|
matched_album=target.album,
|
|
quality=_QUALITIES.get(self.name, _QUALITIES["youtube"]),
|
|
track_count=tracks,
|
|
)
|
|
]
|
|
|
|
def download(
|
|
self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
|
|
) -> DownloadResult:
|
|
for pct in (0.25, 0.5, 0.75, 1.0):
|
|
on_progress(pct)
|
|
return DownloadResult(ok=True, path=dest, track_count=candidate.track_count)
|
|
|
|
|
|
class FakeQobuz(_BaseFake):
|
|
name = "qobuz"
|
|
tier = 0
|
|
|
|
|
|
class FakeSoulseek(_BaseFake):
|
|
name = "soulseek"
|
|
tier = 1
|
|
|
|
|
|
class FakeYouTube(_BaseFake):
|
|
name = "youtube"
|
|
tier = 2
|
|
|
|
|
|
class FailingAdapter(_BaseFake):
|
|
name = "qobuz" # pretends to be a real source that then fails to download
|
|
tier = 0
|
|
|
|
def download(
|
|
self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
|
|
) -> DownloadResult:
|
|
return DownloadResult(ok=False, error="simulated download failure")
|