refactor(domain): unify ETR calculation — both UIs now use calc_etr_str

Before:
  - PyQt: inline ETR using chars-based formula
  - WebUI: Job.estimated_time_remaining using progress-based formula
  (different formulas → different ETR estimates)

After:
  - Both UIs call domain.progress.calc_etr_str(elapsed, done, total)
  - Same formula, same ETR, single source of truth
  - WebUI now stores etr_str on Job and displays it directly
  - Job.estimated_time_remaining property kept for backward compat

domain/progress.py: ProgressTracker class + calc_etr_str function
1053 tests pass.

abogen/domain/progress.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+"""Progress and ETR (estimated time remaining) calculation.
+
+Shared by Web UI and PyQt desktop GUI. Pure math, no UI dependencies.
+"""
+import time
+from dataclasses import dataclass, field
+
+
+@dataclass
+class ProgressTracker:
+    """Tracks character-based progress with ETR calculation.
+
+    Usage:
+        tracker = ProgressTracker(total_chars=50000)
+        # ... as processing occurs:
+        tracker.update(chars_done=5000)
+        print(tracker.etr_str)   # "00:04:30"
+        print(tracker.percent)   # 10
+    """
+    total_chars: int
+    _start_time: float = field(default_factory=time.time, repr=False)
+    _chars_done: int = field(default=0, repr=False)
+
+    def update(self, chars_done: int) -> None:
+        self._chars_done = chars_done
+
+    @property
+    def percent(self) -> int:
+        if self.total_chars <= 0:
+            return 0
+        return min(int(self._chars_done / self.total_chars * 100), 99)
+
+    @property
+    def etr_str(self) -> str:
+        elapsed = time.time() - self._start_time
+        if self._chars_done <= 0 or elapsed <= 0.5:
+            return "Processing..."
+        avg_time_per_char = elapsed / self._chars_done
+        remaining = self.total_chars - self._chars_done
+        if remaining <= 0:
+            return "00:00:00"
+        secs = avg_time_per_char * remaining
+        h = int(secs // 3600)
+        m = int((secs % 3600) // 60)
+        s = int(secs % 60)
+        return f"{h:02d}:{m:02d}:{s:02d}"
+
+
+def calc_etr_str(elapsed: float, done: int, total: int) -> str:
+    """Standalone ETR string calculation (matches PyQt original logic).
+
+    Args:
+        elapsed: seconds since processing started
+        done: items/characters processed so far
+        total: total items/characters to process
+
+    Returns:
+        ETR string like "01:23:45" or "Processing..."
+    """
+    if done <= 0 or elapsed <= 0.5:
+        return "Processing..."
+    avg_time_per_item = elapsed / done
+    remaining = total - done
+    if remaining <= 0:
+        return "00:00:00"
+    secs = avg_time_per_item * remaining
+    h = int(secs // 3600)
+    m = int((secs % 3600) // 60)
+    s = int(secs % 60)
+    return f"{h:02d}:{m:02d}:{s:02d}"

abogen/pyqt/conversion.py

@@ -37,6 +37,7 @@ from abogen.domain.audio_buffer import (
 )
 from abogen.domain.subtitle_generation import process_subtitle_tokens
 from abogen.domain.voice_loader import load_voice_cached
+from abogen.domain.progress import calc_etr_str
 from abogen.domain.metadata_extraction import (
     extract_metadata_and_build_args,
     read_text_for_metadata,
@@ -1266,24 +1267,11 @@ class ConversionThread(QThread):
                             )
 
                             # Calculate ETR based on characters processed
-                            etr_str = "Processing..."
+                            etr_str = calc_etr_str(
-                            chars_done = self.processed_char_count
+                                time.time() - self.etr_start_time,
-                            elapsed = time.time() - self.etr_start_time
+                                self.processed_char_count,
-
+                                self.total_char_count,
-                            # Calculate ETR if enough data is available
+                            )
-                            if (
-                                chars_done > 0 and elapsed > 0.5
-                            ):  # Check elapsed > 0.5 to avoid instability
-                                avg_time_per_char = elapsed / chars_done
-                                remaining = (
-                                    self.total_char_count - self.processed_char_count
-                                )
-                                if remaining > 0:
-                                    secs = avg_time_per_char * remaining
-                                    h = int(secs // 3600)
-                                    m = int((secs % 3600) // 60)
-                                    s = int(secs % 60)
-                                    etr_str = f"{h:02d}:{m:02d}:{s:02d}"
 
                             # Update progress more frequently (after each result)
                             self.progress_updated.emit(percent, etr_str)
@@ -1833,11 +1821,10 @@ class ConversionThread(QThread):
 
                 # Update progress
                 percent = min(int(idx / len(subtitles) * 100), 99)
-                elapsed = time.time() - self.etr_start_time
+                etr_str = calc_etr_str(
-                etr_str = (
+                    time.time() - self.etr_start_time,
-                    "Processing..."
+                    idx,
-                    if elapsed <= 0.5
+                    len(subtitles),
-                    else f"{int(elapsed*(len(subtitles)-idx)/idx)//3600:02d}:{(int(elapsed*(len(subtitles)-idx)/idx)%3600)//60:02d}:{int(elapsed*(len(subtitles)-idx)/idx)%60:02d}"
                 )
                 self.progress_updated.emit(percent, etr_str)
 

abogen/webui/conversion_runner.py

@@ -4,6 +4,7 @@ import json
 import os
 import subprocess
 import sys
+import time
 import traceback
 import gc
 from collections import defaultdict
@@ -112,6 +113,7 @@ from abogen.domain.output_paths import (
 )
 from abogen.domain.device import select_device as _select_device
 from abogen.domain.split_pattern import get_split_pattern
+from abogen.domain.progress import ProgressTracker, calc_etr_str
 from abogen.domain.audio_helpers import (
     build_ffmpeg_command as _build_ffmpeg_command,
     to_float32 as _to_float32,
@@ -408,6 +410,7 @@ def run_conversion_job(job: Job) -> None:
             voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_backend, base_voice_resolved, job.use_gpu)
         processed_chars = 0
         current_time = 0.0
+        etr_start_time = time.time()
         total_chapters = len(extraction.chapters)
         if chunk_groups:
             chunk_groups = {
@@ -508,6 +511,11 @@ def run_conversion_job(job: Job) -> None:
                     job.processed_characters = processed_chars
                     if job.total_characters:
                         job.progress = min(processed_chars / job.total_characters, 0.999)
+                        job.etr_str = calc_etr_str(
+                            time.time() - etr_start_time,
+                            processed_chars,
+                            job.total_characters,
+                        )
                     else:
                         job.progress = 0.0 if processed_chars == 0 else 0.999
 

abogen/webui/service.py

@@ -137,6 +137,7 @@ class Job:
     progress: float = 0.0
     total_characters: int = 0
     processed_characters: int = 0
+    etr_str: str = ""
     logs: List[JobLog] = field(default_factory=list)
     error: Optional[str] = None
     result: JobResult = field(default_factory=JobResult)
@@ -168,20 +169,25 @@ class Job:
     @property
     def estimated_time_remaining(self) -> Optional[float]:
         """
-        Returns the estimated seconds remaining based on current progress and elapsed time.
+        Returns the estimated seconds remaining.
-        Returns None if the job hasn't started, is finished, or progress is 0.
+        Uses the same calc_etr_str from domain/progress.py as the PyQt desktop GUI.
         """
-        if self.status != JobStatus.RUNNING or not self.started_at or self.progress <= 0:
+        from abogen.domain.progress import calc_etr_str
+
+        if self.status != JobStatus.RUNNING or not self.started_at or self.total_characters <= 0:
             return None
-        
+
         elapsed = time.time() - self.started_at
         if elapsed <= 0:
             return None
-            
+
-        # Estimate total time based on current progress
+        etr = calc_etr_str(elapsed, self.processed_characters, self.total_characters)
-        total_estimated = elapsed / self.progress
+        if etr == "Processing...":
-        remaining = total_estimated - elapsed
+            return None
-        return max(0.0, remaining)
+
+        # Parse "HH:MM:SS" back to seconds for backward compatibility
+        parts = etr.split(":")
+        return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2])
 
     def add_log(self, message: str, level: str = "info") -> None:
         entry = JobLog(timestamp=time.time(), message=message, level=level)
@@ -200,6 +206,7 @@ class Job:
             "progress": self.progress,
             "total_characters": self.total_characters,
             "processed_characters": self.processed_characters,
+            "etr_str": self.etr_str,
             "error": self.error,
             "logs": [log.__dict__ for log in self.logs],
             "result": {

abogen/webui/templates/partials/jobs.html

@@ -28,8 +28,8 @@
             </div>
             <div class="job-card__progress-meta">
               <small>{{ progress_value }}% · {{ job.processed_characters }} / {{ job.total_characters or '—' }}</small>
-              {% if job.estimated_time_remaining %}
+              {% if job.etr_str and job.etr_str != 'Processing...' %}
-                <small class="job-card__eta">~{{ job.estimated_time_remaining | durationformat }} remaining</small>
+                <small class="job-card__eta">~{{ job.etr_str }} remaining</small>
               {% endif %}
             </div>
           </div>

tests/test_domain_progress.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import time
+from unittest.mock import patch
+
+from abogen.domain.progress import ProgressTracker, calc_etr_str
+
+
+class TestCalcEtrStr:
+    def test_returns_processing_when_not_started(self):
+        assert calc_etr_str(0.0, 0, 100) == "Processing..."
+
+    def test_returns_processing_when_elapsed_too_short(self):
+        assert calc_etr_str(0.3, 50, 100) == "Processing..."
+
+    def test_returns_processing_when_done_zero(self):
+        assert calc_etr_str(2.0, 0, 100) == "Processing..."
+
+    def test_returns_zero_when_complete(self):
+        assert calc_etr_str(10.0, 100, 100) == "00:00:00"
+
+    def test_returns_zero_when_overcomplete(self):
+        assert calc_etr_str(10.0, 150, 100) == "00:00:00"
+
+    def test_half_done(self):
+        # 10s for 50 chars -> 10s remaining -> 00:00:10
+        assert calc_etr_str(10.0, 50, 100) == "00:00:10"
+
+    def test_one_third_done_large(self):
+        # 100s for 1000 chars out of 3000 -> 200s remaining
+        assert calc_etr_str(100.0, 1000, 3000) == "00:03:20"
+
+    def test_hours_format(self):
+        # 3600s for 1000 chars out of 4000 -> 3 * 3600 = 10800s remaining
+        assert calc_etr_str(3600.0, 1000, 4000) == "03:00:00"
+
+    def test_minutes_and_seconds(self):
+        # 60s for 100 chars out of 200 -> 60s remaining
+        assert calc_etr_str(60.0, 100, 200) == "00:01:00"
+
+
+class TestProgressTracker:
+    def test_percent_at_zero(self):
+        t = ProgressTracker(total_chars=1000)
+        assert t.percent == 0
+
+    def test_percent_half(self):
+        t = ProgressTracker(total_chars=1000)
+        t.update(500)
+        assert t.percent == 50
+
+    def test_percent_capped_at_99(self):
+        t = ProgressTracker(total_chars=100)
+        t.update(100)
+        assert t.percent == 99  # matches original behavior
+
+    def test_etr_processing_at_start(self):
+        t = ProgressTracker(total_chars=1000)
+        t.update(0)
+        assert t.etr_str == "Processing..."
+
+    def test_etr_computes_correctly(self):
+        t = ProgressTracker(total_chars=200)
+        with patch("abogen.domain.progress.time") as mock_time:
+            mock_time.time.return_value = t._start_time + 2.0
+            t.update(100)
+            assert t.etr_str == "00:00:02"
+
+    def test_zero_total_chars(self):
+        t = ProgressTracker(total_chars=0)
+        assert t.percent == 0