diff --git a/app/library/Events.py b/app/library/Events.py index 3a514a84..2b6e3b78 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -34,6 +34,7 @@ class Events: ITEM_ADDED: str = "item_added" ITEM_UPDATED: str = "item_updated" + ITEM_PROGRESS: str = "item_progress" ITEM_COMPLETED: str = "item_completed" ITEM_CANCELLED: str = "item_cancelled" ITEM_DELETED: str = "item_deleted" @@ -86,6 +87,7 @@ class Events: Events.LOG_SUCCESS, Events.ITEM_ADDED, Events.ITEM_UPDATED, + Events.ITEM_PROGRESS, Events.ITEM_CANCELLED, Events.ITEM_DELETED, Events.ITEM_BULK_DELETED, @@ -103,7 +105,7 @@ class Events: list: The list of debug events. """ - return [Events.ITEM_UPDATED] + return [Events.ITEM_UPDATED, Events.ITEM_PROGRESS] @dataclass(kw_only=True) diff --git a/app/library/downloads/status_tracker.py b/app/library/downloads/status_tracker.py index 5a29a77c..704a9a17 100644 --- a/app/library/downloads/status_tracker.py +++ b/app/library/downloads/status_tracker.py @@ -56,6 +56,36 @@ class StatusTracker: self._terminator_sent: bool = False self._candidate_filepath: Path | None = None self.update_task: asyncio.Task | None = None + self._last_progress_time: float = 0.0 + self._pending_progress: bool = False + self._progress_interval: float = 0.5 + + def _progress_payload(self) -> dict: + return { + "_id": self.info._id, + "status": self.info.status, + "percent": self.info.percent, + "speed": self.info.speed, + "eta": self.info.eta, + "downloaded_bytes": self.info.downloaded_bytes, + "total_bytes": self.info.total_bytes, + "msg": self.info.msg, + } + + def _emit_progress(self) -> None: + now = time.monotonic() + if (now - self._last_progress_time) >= self._progress_interval: + self._notify.emit(Events.ITEM_PROGRESS, data=self._progress_payload()) + self._last_progress_time = now + self._pending_progress = False + else: + self._pending_progress = True + + def _flush_progress(self) -> None: + if self._pending_progress: + self._notify.emit(Events.ITEM_PROGRESS, data=self._progress_payload()) + self._pending_progress = False + self._last_progress_time = time.monotonic() async def _finalize_file(self, filepath: Path) -> None: """ @@ -184,6 +214,7 @@ class StatusTracker: self.tmpfilename = status.get("tmpfilename") + old_status = self.info.status self.info.status = status.get("status", self.info.status) if "download_skipped" in status: self.info.download_skipped = bool(status.get("download_skipped")) @@ -235,7 +266,11 @@ class StatusTracker: await self._finalize_file(Path(final_name)) self.info.status = "finished" - self._notify.emit(Events.ITEM_UPDATED, data=self.info) + if self.info.status != old_status or self.final_update: + self._flush_progress() + self._notify.emit(Events.ITEM_UPDATED, data=self.info) + else: + self._emit_progress() async def progress_update(self) -> None: """ @@ -248,9 +283,11 @@ class StatusTracker: self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get) status = await self.update_task if status is None or isinstance(status, Terminator): + self._flush_progress() return await self.process_status_update(status) except (asyncio.CancelledError, OSError, FileNotFoundError, EOFError, BrokenPipeError, ConnectionError): + self._flush_progress() return async def drain_queue(self, max_iterations: int = 50) -> None: @@ -295,6 +332,8 @@ class StatusTracker: except (queue.Empty, BrokenPipeError, ConnectionRefusedError, EOFError, OSError): continue + self._flush_progress() + def cancel_update_task(self) -> None: """Cancel the progress update task if it's running.""" try: diff --git a/app/tests/test_download.py b/app/tests/test_download.py index 22228ce1..f5440d0e 100644 --- a/app/tests/test_download.py +++ b/app/tests/test_download.py @@ -1211,6 +1211,92 @@ class TestStatusTracker: assert 1 == len(queue.items), "Should add terminator to queue" assert isinstance(queue.items[0], Terminator), "Should add Terminator instance" + @pytest.mark.asyncio + async def test_progress_emits_item_progress(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + st.info.status = "downloading" + calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: calls.append((a, kw))) + + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 50, "total_bytes": 100} + ) + + progress_calls = [c for c in calls if c[0][0] == Events.ITEM_PROGRESS] + updated_calls = [c for c in calls if c[0][0] == Events.ITEM_UPDATED] + assert len(progress_calls) == 1 + assert len(updated_calls) == 0 + payload = progress_calls[0][1]["data"] + assert payload["_id"] == st.info._id + assert payload["percent"] == 50.0 + assert "options" not in payload + + @pytest.mark.asyncio + async def test_status_change_emits_item_updated(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + st.info.status = "started" + calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: calls.append((a, kw))) + + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 10, "total_bytes": 100} + ) + + updated_calls = [c for c in calls if c[0][0] == Events.ITEM_UPDATED] + assert len(updated_calls) == 1 + assert updated_calls[0][1]["data"] is st.info + + @pytest.mark.asyncio + async def test_progress_throttled(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + st.info.status = "downloading" + st._progress_interval = 0.5 + calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: calls.append((a, kw))) + + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 10, "total_bytes": 100} + ) + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 20, "total_bytes": 100} + ) + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 30, "total_bytes": 100} + ) + + progress_calls = [c for c in calls if c[0][0] == Events.ITEM_PROGRESS] + assert len(progress_calls) == 1, "Rapid ticks should be throttled to one emission" + assert st._pending_progress is True + + @pytest.mark.asyncio + async def test_flush_on_status_change(self, mock_config: dict) -> None: + st = StatusTracker(**mock_config) + st.info.status = "downloading" + st._progress_interval = 10.0 + calls: list = [] + st._notify = Mock() + st._notify.emit = Mock(side_effect=lambda *a, **kw: calls.append((a, kw))) + + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 10, "total_bytes": 100} + ) + await st.process_status_update( + {"id": "test-id", "status": "downloading", "downloaded_bytes": 50, "total_bytes": 100} + ) + + assert st._pending_progress is True + + await st.process_status_update({"id": "test-id", "status": "error", "error": "fail"}) + + progress_calls = [c for c in calls if c[0][0] == Events.ITEM_PROGRESS] + updated_calls = [c for c in calls if c[0][0] == Events.ITEM_UPDATED] + assert len(progress_calls) == 2, "Pending progress should be flushed before status change" + assert len(updated_calls) == 1 + assert st._pending_progress is False + class TestQueueManager: @staticmethod diff --git a/ui/app/composables/useAppSocket.ts b/ui/app/composables/useAppSocket.ts index 6d562565..36bd860e 100644 --- a/ui/app/composables/useAppSocket.ts +++ b/ui/app/composables/useAppSocket.ts @@ -343,6 +343,15 @@ on('item_updated', (data: WSEP['item_updated']) => { } }); +on('item_progress', (data: WSEP['item_progress']) => { + const queueState = getQueueState(); + const id = data.data._id; + + if (true === queueState.has(id)) { + queueState.patch(id, data.data as Partial); + } +}); + on('item_moved', (data: WSEP['item_moved']) => { const queueState = getQueueState(); const to = data.data.to; diff --git a/ui/app/composables/useQueueState.ts b/ui/app/composables/useQueueState.ts index 62f76417..27f12736 100644 --- a/ui/app/composables/useQueueState.ts +++ b/ui/app/composables/useQueueState.ts @@ -21,6 +21,12 @@ const update = (key: KeyType, value: StoreItem): void => { state.queue[key] = value; }; +const patch = (key: KeyType, fields: Partial): void => { + if (state.queue[key]) { + Object.assign(state.queue[key], fields); + } +}; + const remove = (key: KeyType): void => { if (!state.queue[key]) { return; @@ -227,6 +233,7 @@ const queueStateApi = proxyRefs({ ...toRefs(state), add, update, + patch, remove, get, has, diff --git a/ui/app/types/sockets.d.ts b/ui/app/types/sockets.d.ts index 23851cf6..00c6c715 100644 --- a/ui/app/types/sockets.d.ts +++ b/ui/app/types/sockets.d.ts @@ -1,4 +1,15 @@ -import type { StoreItem } from './store'; +import type { ItemStatus, StoreItem } from './store'; + +export type ItemProgress = { + _id: string; + status: ItemStatus; + percent?: number | null; + speed?: string | null; + eta?: string | null; + downloaded_bytes?: number | null; + total_bytes?: number | null; + msg?: string | null; +}; export type Event = { id: string; @@ -23,6 +34,7 @@ export type WSEP = { connected: EventPayload<{ sid: string }>; item_added: EventPayload; item_updated: EventPayload; + item_progress: EventPayload; item_cancelled: EventPayload; item_deleted: EventPayload; item_bulk_deleted: EventPayload<{ count: number; status?: string; ids?: string[] }>;