C5: Migrate HiFi to engine.worker
Same pattern as C2/C3/C4. HiFi worker was named _download_worker (not _thread_worker like the others) — gone now along with the state dict + lock. Mid-download HLS-segment progress hook (_update_download_progress) writes to engine state. Pinning tests updated. Suite still green (318 download tests).
This commit is contained in:
Broque Thomas
parent
7944568c5c
commit
27a97f8af6
2 changed files with 144 additions and 179 deletions
|
|
@ -110,9 +110,7 @@ class HiFiClient:
|
|||
'Accept': 'application/json',
|
||||
})
|
||||
|
||||
self.active_downloads: Dict[str, Dict[str, Any]] = {}
|
||||
self._download_lock = threading.Lock()
|
||||
|
||||
self._engine = None
|
||||
self.shutdown_check = None
|
||||
|
||||
self._last_api_call = 0
|
||||
|
|
@ -125,6 +123,9 @@ class HiFiClient:
|
|||
def set_shutdown_check(self, check_callable):
|
||||
self.shutdown_check = check_callable
|
||||
|
||||
def set_engine(self, engine):
|
||||
self._engine = engine
|
||||
|
||||
def _load_instances_from_db(self):
|
||||
try:
|
||||
from database.music_database import get_database
|
||||
|
|
@ -571,71 +572,31 @@ class HiFiClient:
|
|||
)
|
||||
|
||||
async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]:
|
||||
try:
|
||||
if '||' not in filename:
|
||||
logger.error(f"Invalid filename format: {filename}")
|
||||
return None
|
||||
|
||||
track_id_str, display_name = filename.split('||', 1)
|
||||
try:
|
||||
track_id = int(track_id_str)
|
||||
except ValueError:
|
||||
logger.error(f"Invalid track ID: {track_id_str}")
|
||||
return None
|
||||
|
||||
download_id = str(uuid.uuid4())
|
||||
|
||||
with self._download_lock:
|
||||
self.active_downloads[download_id] = {
|
||||
'id': download_id,
|
||||
'filename': filename,
|
||||
'username': 'hifi',
|
||||
'state': 'Initializing',
|
||||
'progress': 0.0,
|
||||
'size': 0,
|
||||
'transferred': 0,
|
||||
'speed': 0,
|
||||
'time_remaining': None,
|
||||
'track_id': track_id,
|
||||
'display_name': display_name,
|
||||
'file_path': None,
|
||||
}
|
||||
|
||||
thread = threading.Thread(
|
||||
target=self._download_worker,
|
||||
args=(download_id, track_id, display_name),
|
||||
daemon=True,
|
||||
)
|
||||
thread.start()
|
||||
|
||||
return download_id
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to start HiFi download: {e}")
|
||||
if '||' not in filename:
|
||||
logger.error(f"Invalid filename format: {filename}")
|
||||
return None
|
||||
if self._engine is None:
|
||||
logger.error("HiFi client has no engine reference — cannot dispatch download")
|
||||
return None
|
||||
|
||||
def _download_worker(self, download_id: str, track_id: int, display_name: str):
|
||||
track_id_str, display_name = filename.split('||', 1)
|
||||
try:
|
||||
with self._download_lock:
|
||||
if download_id in self.active_downloads:
|
||||
self.active_downloads[download_id]['state'] = 'InProgress, Downloading'
|
||||
track_id = int(track_id_str)
|
||||
except ValueError:
|
||||
logger.error(f"Invalid track ID: {track_id_str}")
|
||||
return None
|
||||
|
||||
file_path = self._download_sync(download_id, track_id, display_name)
|
||||
|
||||
with self._download_lock:
|
||||
if download_id in self.active_downloads:
|
||||
if file_path:
|
||||
self.active_downloads[download_id]['state'] = 'Completed, Succeeded'
|
||||
self.active_downloads[download_id]['progress'] = 100.0
|
||||
self.active_downloads[download_id]['file_path'] = file_path
|
||||
else:
|
||||
self.active_downloads[download_id]['state'] = 'Errored'
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"HiFi download worker failed for {download_id}: {e}")
|
||||
with self._download_lock:
|
||||
if download_id in self.active_downloads:
|
||||
self.active_downloads[download_id]['state'] = 'Errored'
|
||||
return self._engine.worker.dispatch(
|
||||
source_name='hifi',
|
||||
target_id=track_id,
|
||||
display_name=display_name,
|
||||
original_filename=filename,
|
||||
impl_callable=self._download_sync,
|
||||
extra_record_fields={
|
||||
'track_id': track_id,
|
||||
'display_name': display_name,
|
||||
},
|
||||
)
|
||||
|
||||
def _download_sync(self, download_id: str, track_id: int, display_name: str) -> Optional[str]:
|
||||
quality_key = config_manager.get('hifi_download.quality', 'lossless')
|
||||
|
|
@ -676,9 +637,8 @@ class HiFiClient:
|
|||
speed_start = time.time()
|
||||
segments_completed = 0
|
||||
|
||||
with self._download_lock:
|
||||
if download_id in self.active_downloads:
|
||||
self.active_downloads[download_id]['size'] = 0
|
||||
if self._engine is not None:
|
||||
self._engine.update_record('hifi', download_id, {'size': 0})
|
||||
|
||||
with intermediate_path.open('wb') as output_file:
|
||||
if init_uri:
|
||||
|
|
@ -777,80 +737,77 @@ class HiFiClient:
|
|||
def _update_download_progress(self, download_id: str, downloaded: int,
|
||||
segments_completed: int, total_segments: int,
|
||||
speed_start: float):
|
||||
with self._download_lock:
|
||||
if download_id not in self.active_downloads:
|
||||
return
|
||||
info = self.active_downloads[download_id]
|
||||
info['transferred'] = downloaded
|
||||
if self._engine is None:
|
||||
return
|
||||
record = self._engine.get_record('hifi', download_id)
|
||||
if record is None:
|
||||
return
|
||||
|
||||
now = time.time()
|
||||
elapsed_total = now - speed_start
|
||||
speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0
|
||||
info['speed'] = speed
|
||||
now = time.time()
|
||||
elapsed_total = now - speed_start
|
||||
speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0
|
||||
|
||||
if total_segments > 0:
|
||||
progress = (segments_completed / total_segments) * 100
|
||||
info['progress'] = round(min(progress, 99.9), 1)
|
||||
progress = record.get('progress', 0.0)
|
||||
if total_segments > 0:
|
||||
progress = round(min((segments_completed / total_segments) * 100, 99.9), 1)
|
||||
|
||||
time_remaining = None
|
||||
if speed > 0:
|
||||
remaining_bytes = downloaded * (total_segments / max(segments_completed, 1)) - downloaded
|
||||
if remaining_bytes > 0:
|
||||
time_remaining = int(remaining_bytes / speed)
|
||||
info['time_remaining'] = time_remaining
|
||||
time_remaining = None
|
||||
if speed > 0:
|
||||
remaining_bytes = downloaded * (total_segments / max(segments_completed, 1)) - downloaded
|
||||
if remaining_bytes > 0:
|
||||
time_remaining = int(remaining_bytes / speed)
|
||||
|
||||
self._engine.update_record('hifi', download_id, {
|
||||
'transferred': downloaded,
|
||||
'speed': speed,
|
||||
'progress': progress,
|
||||
'time_remaining': time_remaining,
|
||||
})
|
||||
|
||||
def _record_to_status(self, record):
|
||||
return DownloadStatus(
|
||||
id=record['id'],
|
||||
filename=record['filename'],
|
||||
username=record['username'],
|
||||
state=record['state'],
|
||||
progress=record['progress'],
|
||||
size=record.get('size', 0),
|
||||
transferred=record.get('transferred', 0),
|
||||
speed=record.get('speed', 0),
|
||||
time_remaining=record.get('time_remaining'),
|
||||
file_path=record.get('file_path'),
|
||||
)
|
||||
|
||||
async def get_all_downloads(self) -> List[DownloadStatus]:
|
||||
statuses = []
|
||||
with self._download_lock:
|
||||
for _dl_id, info in self.active_downloads.items():
|
||||
statuses.append(DownloadStatus(
|
||||
id=info['id'],
|
||||
filename=info['filename'],
|
||||
username=info['username'],
|
||||
state=info['state'],
|
||||
progress=info['progress'],
|
||||
size=info['size'],
|
||||
transferred=info['transferred'],
|
||||
speed=info['speed'],
|
||||
time_remaining=info.get('time_remaining'),
|
||||
file_path=info.get('file_path'),
|
||||
))
|
||||
return statuses
|
||||
if self._engine is None:
|
||||
return []
|
||||
return [
|
||||
self._record_to_status(record)
|
||||
for record in self._engine.iter_records_for_source('hifi')
|
||||
]
|
||||
|
||||
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
|
||||
with self._download_lock:
|
||||
info = self.active_downloads.get(download_id)
|
||||
if not info:
|
||||
return None
|
||||
return DownloadStatus(
|
||||
id=info['id'],
|
||||
filename=info['filename'],
|
||||
username=info['username'],
|
||||
state=info['state'],
|
||||
progress=info['progress'],
|
||||
size=info['size'],
|
||||
transferred=info['transferred'],
|
||||
speed=info['speed'],
|
||||
time_remaining=info.get('time_remaining'),
|
||||
file_path=info.get('file_path'),
|
||||
)
|
||||
if self._engine is None:
|
||||
return None
|
||||
record = self._engine.get_record('hifi', download_id)
|
||||
return self._record_to_status(record) if record is not None else None
|
||||
|
||||
async def cancel_download(self, download_id: str, username: str = None,
|
||||
remove: bool = False) -> bool:
|
||||
with self._download_lock:
|
||||
if download_id not in self.active_downloads:
|
||||
return False
|
||||
self.active_downloads[download_id]['state'] = 'Cancelled'
|
||||
if remove:
|
||||
del self.active_downloads[download_id]
|
||||
if self._engine is None:
|
||||
return False
|
||||
if self._engine.get_record('hifi', download_id) is None:
|
||||
return False
|
||||
self._engine.update_record('hifi', download_id, {'state': 'Cancelled'})
|
||||
if remove:
|
||||
self._engine.remove_record('hifi', download_id)
|
||||
return True
|
||||
|
||||
async def clear_all_completed_downloads(self) -> bool:
|
||||
with self._download_lock:
|
||||
to_remove = [
|
||||
did for did, info in self.active_downloads.items()
|
||||
if info.get('state', '') in ('Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted')
|
||||
]
|
||||
for did in to_remove:
|
||||
del self.active_downloads[did]
|
||||
if self._engine is None:
|
||||
return True
|
||||
terminal = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'}
|
||||
for record in list(self._engine.iter_records_for_source('hifi')):
|
||||
if record.get('state') in terminal:
|
||||
self._engine.remove_record('hifi', record['id'])
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -1,10 +1,4 @@
|
|||
"""Phase A pinning tests for HiFiClient's download lifecycle.
|
||||
|
||||
HiFi uses public hifi-api instances backed by Tidal-sourced metadata.
|
||||
Same int track_id + thread-worker + state-dict pattern as Tidal/Qobuz,
|
||||
EXCEPT the worker method is named `_download_worker` (no `_thread_`).
|
||||
Engine refactor must preserve the worker target signature.
|
||||
"""
|
||||
"""Phase A pinning tests for HiFiClient — UPDATED for Phase C5."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -15,6 +9,7 @@ from unittest.mock import patch
|
|||
|
||||
import pytest
|
||||
|
||||
from core.download_engine import DownloadEngine
|
||||
from core.hifi_client import HiFiClient
|
||||
|
||||
|
||||
|
|
@ -27,65 +22,78 @@ def _run_async(coro):
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def hifi_client():
|
||||
def hifi_client_with_engine():
|
||||
client = HiFiClient.__new__(HiFiClient)
|
||||
client.download_path = Path('./test_hifi_downloads')
|
||||
client.shutdown_check = None
|
||||
client.active_downloads = {}
|
||||
client._download_lock = threading.Lock()
|
||||
return client
|
||||
client._engine = None
|
||||
engine = DownloadEngine()
|
||||
client.set_engine(engine)
|
||||
return client, engine
|
||||
|
||||
|
||||
def test_download_returns_none_for_invalid_filename_format(hifi_client):
|
||||
result = _run_async(hifi_client.download('hifi', 'no-separator', 0))
|
||||
def test_download_returns_none_for_invalid_filename_format(hifi_client_with_engine):
|
||||
client, _ = hifi_client_with_engine
|
||||
result = _run_async(client.download('hifi', 'no-separator', 0))
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_download_returns_none_for_non_integer_track_id(hifi_client):
|
||||
result = _run_async(hifi_client.download('hifi', 'not-int||title', 0))
|
||||
def test_download_returns_none_for_non_integer_track_id(hifi_client_with_engine):
|
||||
client, _ = hifi_client_with_engine
|
||||
result = _run_async(client.download('hifi', 'not-int||title', 0))
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_download_returns_uuid_for_valid_filename(hifi_client):
|
||||
with patch('core.hifi_client.threading.Thread') as fake:
|
||||
fake.return_value.start = lambda: None
|
||||
result = _run_async(hifi_client.download('hifi', '12345||Some Song', 0))
|
||||
def test_download_returns_none_when_engine_not_wired():
|
||||
client = HiFiClient.__new__(HiFiClient)
|
||||
client._engine = None
|
||||
result = _run_async(client.download('hifi', '12345||x', 0))
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_download_returns_uuid_for_valid_filename(hifi_client_with_engine):
|
||||
client, _ = hifi_client_with_engine
|
||||
with patch.object(client, '_download_sync', return_value='/tmp/x.flac'):
|
||||
result = _run_async(client.download('hifi', '12345||Some Song', 0))
|
||||
assert result is not None
|
||||
assert len(result) == 36
|
||||
|
||||
|
||||
def test_download_populates_active_downloads_with_initial_state(hifi_client):
|
||||
with patch('core.hifi_client.threading.Thread') as fake:
|
||||
fake.return_value.start = lambda: None
|
||||
download_id = _run_async(hifi_client.download('hifi', '999||My HiFi Song', 0))
|
||||
def test_download_populates_engine_record_with_initial_state(hifi_client_with_engine):
|
||||
client, engine = hifi_client_with_engine
|
||||
started = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
record = hifi_client.active_downloads[download_id]
|
||||
assert record['id'] == download_id
|
||||
assert record['filename'] == '999||My HiFi Song'
|
||||
assert record['username'] == 'hifi'
|
||||
assert record['state'] == 'Initializing'
|
||||
assert record['track_id'] == 999
|
||||
assert record['display_name'] == 'My HiFi Song'
|
||||
def slow_impl(*args, **kwargs):
|
||||
started.set()
|
||||
release.wait(timeout=1.0)
|
||||
return '/tmp/done.flac'
|
||||
|
||||
with patch.object(client, '_download_sync', side_effect=slow_impl):
|
||||
download_id = _run_async(client.download('hifi', '999||My HiFi Song', 0))
|
||||
started.wait(timeout=1.0)
|
||||
record = engine.get_record('hifi', download_id)
|
||||
assert record['filename'] == '999||My HiFi Song'
|
||||
assert record['username'] == 'hifi'
|
||||
assert record['track_id'] == 999
|
||||
assert record['display_name'] == 'My HiFi Song'
|
||||
release.set()
|
||||
|
||||
|
||||
def test_download_spawns_daemon_thread_targeting_download_worker(hifi_client):
|
||||
"""Pinning: target is `_download_worker` (NOT `_thread_worker` like
|
||||
Tidal/Qobuz). Engine refactor's plugin contract must accommodate
|
||||
this naming variance OR force a rename — pinned here so the
|
||||
decision is conscious."""
|
||||
captured_kwargs = {}
|
||||
def test_get_all_downloads_reads_engine_records(hifi_client_with_engine):
|
||||
client, engine = hifi_client_with_engine
|
||||
engine.add_record('hifi', 'dl-1', {
|
||||
'id': 'dl-1', 'filename': '111||A', 'username': 'hifi',
|
||||
'state': 'InProgress, Downloading', 'progress': 50.0,
|
||||
})
|
||||
result = _run_async(client.get_all_downloads())
|
||||
assert len(result) == 1
|
||||
assert result[0].id == 'dl-1'
|
||||
|
||||
def capture_thread(*args, **kwargs):
|
||||
captured_kwargs.update(kwargs)
|
||||
return type('FakeThread', (), {'start': lambda self: None})()
|
||||
|
||||
with patch('core.hifi_client.threading.Thread', side_effect=capture_thread):
|
||||
_run_async(hifi_client.download('hifi', '777||Title', 0))
|
||||
|
||||
assert captured_kwargs.get('daemon') is True
|
||||
assert captured_kwargs.get('target') == hifi_client._download_worker
|
||||
args = captured_kwargs.get('args', ())
|
||||
# HiFi's worker signature: (download_id, track_id, display_name) — 3 args, not 4
|
||||
assert len(args) == 3
|
||||
assert args[1] == 777
|
||||
assert args[2] == 'Title'
|
||||
def test_cancel_download_marks_cancelled(hifi_client_with_engine):
|
||||
client, engine = hifi_client_with_engine
|
||||
engine.add_record('hifi', 'dl-1', {'id': 'dl-1', 'state': 'InProgress, Downloading'})
|
||||
ok = _run_async(client.cancel_download('dl-1', None, remove=False))
|
||||
assert ok is True
|
||||
assert engine.get_record('hifi', 'dl-1')['state'] == 'Cancelled'
|
||||
|
|
|
|||
Loading…
Reference in a new issue