mirror of
https://github.com/denizsafak/abogen.git
synced 2026-09-20 11:40:57 +02:00
feat: application layer models and ports for conversion unification
- application/conversion_models.py: SegmentPlan, ChapterPlan, ConversionPlan, OutputLayout, IntroOutroSpec - application/conversion_request.py: ConversionRequest (normalized input) - application/conversion_result.py: ConversionResult, ConversionError (normalized output) - application/conversion_ports.py: protocols (ConversionEvents, PipelineProvider, VoiceResolver, SubtitleWriter, AudioSink) These are pure data models and interfaces. No implementation yet.
This commit is contained in:
Artem Akymenko
@@ -0,0 +1,8 @@
|
||||
"""Application layer for conversion flow unification.
|
||||
|
||||
This package contains the application-level orchestration logic
|
||||
that bridges UI adapters (PyQt, WebUI) with domain functions.
|
||||
|
||||
The main entry point is ConversionService.run() which coordinates
|
||||
planning, execution, and finalization of a conversion job.
|
||||
"""
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Core models for conversion planning.
|
||||
|
||||
These dataclasses represent the structured plan for a conversion job.
|
||||
They are UI-agnostic and describe WHAT to convert, not HOW to do it.
|
||||
|
||||
The planning flow:
|
||||
ConversionRequest -> ConversionPlan -> ConversionResult
|
||||
|
||||
ConversionPlan contains:
|
||||
- ChapterPlan[]: chapters with their segments
|
||||
- SegmentPlan[]: individual text segments with voice specs
|
||||
- OutputLayout: where to write outputs
|
||||
- IntroOutroSpec: optional intro/outro
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class SegmentPlan:
|
||||
"""A single text segment with its voice specification.
|
||||
|
||||
This is the unified model for:
|
||||
- Regular chapter body text
|
||||
- PyQt voice markers (<<VOICE:F1>>)
|
||||
- WebUI chunks with per-chunk voice/speaker
|
||||
- Intro/outro text
|
||||
- Chapter headings
|
||||
"""
|
||||
|
||||
text: str
|
||||
voice_spec: str
|
||||
kind: str = "body" # intro, heading, body, outro
|
||||
speaker_id: str = "narrator"
|
||||
chunk_id: Optional[str] = None
|
||||
chunk_index: Optional[int] = None
|
||||
level: Optional[str] = None # chunk level (paragraph, sentence, etc.)
|
||||
source: str = "chapter" # chapter, voice_marker, chunk
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChapterPlan:
|
||||
"""A chapter with its metadata and segments."""
|
||||
|
||||
index: int
|
||||
title: str
|
||||
original_title: str
|
||||
body_text: str
|
||||
segments: List[SegmentPlan]
|
||||
voice_spec: str # default voice for this chapter
|
||||
|
||||
|
||||
@dataclass
|
||||
class OutputLayout:
|
||||
"""Resolved output paths for a conversion job."""
|
||||
|
||||
parent_dir: Path
|
||||
merged_path: Optional[Path] = None
|
||||
chapter_dir: Optional[Path] = None
|
||||
project_root: Optional[Path] = None
|
||||
audio_dir: Optional[Path] = None
|
||||
subtitle_dir: Optional[Path] = None
|
||||
metadata_dir: Optional[Path] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class IntroOutroSpec:
|
||||
"""Intro/outro specification with resolved text and voice."""
|
||||
|
||||
enabled: bool = False
|
||||
text: str = ""
|
||||
voice_spec: str = ""
|
||||
kind: str = "intro" # intro or outro
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionPlan:
|
||||
"""Complete plan for a conversion job.
|
||||
|
||||
This is the output of the planning phase and input to the executor.
|
||||
"""
|
||||
|
||||
request: Any # ConversionRequest (forward reference to avoid circular import)
|
||||
metadata: Dict[str, Any]
|
||||
chapters: List[ChapterPlan]
|
||||
intro: Optional[IntroOutroSpec] = None
|
||||
outro: Optional[IntroOutroSpec] = None
|
||||
output_layout: Optional[OutputLayout] = None
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Ports / interfaces for the conversion service.
|
||||
|
||||
These protocols define how the conversion service communicates with
|
||||
the outside world (UI, TTS backends, voice resolvers).
|
||||
|
||||
The service ONLY depends on these interfaces, never on concrete
|
||||
implementations (PyQt signals, Flask Job, etc.).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, List, Optional, Protocol, runtime_checkable
|
||||
|
||||
|
||||
class ConversionEvents(Protocol):
|
||||
"""UI-specific actions the conversion service delegates back to the caller.
|
||||
|
||||
Implementations:
|
||||
- PyQt: emits signals (log_updated, progress_updated, etc.)
|
||||
- WebUI: updates Job attributes (job.add_log, job.progress, etc.)
|
||||
"""
|
||||
|
||||
def log(self, message: str, level: str = "info") -> None:
|
||||
"""Log a message to the UI."""
|
||||
...
|
||||
|
||||
def progress(self, processed: int, total: int, etr: str) -> None:
|
||||
"""Update progress display."""
|
||||
...
|
||||
|
||||
def check_cancelled(self) -> None:
|
||||
"""Check if conversion was cancelled.
|
||||
|
||||
Should raise ConversionCancelled (or UI-specific exception)
|
||||
if cancellation is requested. Normal return means "continue".
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class PipelineProvider(Protocol):
|
||||
"""Provides access to TTS backends (Kokoro, SuperTonic, etc.).
|
||||
|
||||
Implementations:
|
||||
- PyQt: wraps self.backend (single pipeline)
|
||||
- WebUI: wraps PipelinePool (multi-provider)
|
||||
"""
|
||||
|
||||
def get(self, provider: str, language: str, use_gpu: bool) -> Any:
|
||||
"""Get a TTS backend instance."""
|
||||
...
|
||||
|
||||
def dispose_all(self) -> None:
|
||||
"""Dispose all backend resources."""
|
||||
...
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class ResolvedVoice:
|
||||
"""A resolved voice ready for TTS synthesis."""
|
||||
|
||||
provider: str
|
||||
resolved_spec: str
|
||||
voice: Any # loaded voice tensor or name
|
||||
speed: float
|
||||
supertonic_steps: int
|
||||
|
||||
|
||||
class VoiceResolver(Protocol):
|
||||
"""Resolves voice specs into loaded voice objects.
|
||||
|
||||
Implementations:
|
||||
- PyQt: wraps load_voice_cached + VoiceCache
|
||||
- WebUI: wraps resolve_voice_choice + PipelinePool + VoiceCache
|
||||
"""
|
||||
|
||||
def resolve(self, voice_spec: str) -> ResolvedVoice:
|
||||
"""Resolve a voice spec into a loaded voice."""
|
||||
...
|
||||
|
||||
|
||||
class SubtitleWriter(Protocol):
|
||||
"""Writes subtitle entries to a file."""
|
||||
|
||||
def open(self) -> None:
|
||||
"""Open the subtitle file for writing."""
|
||||
...
|
||||
|
||||
def write_entry(self, start: float, end: float, text: str) -> None:
|
||||
"""Write a single subtitle entry."""
|
||||
...
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the subtitle file."""
|
||||
...
|
||||
|
||||
|
||||
class AudioSink(Protocol):
|
||||
"""Writes audio data to a file."""
|
||||
|
||||
def write(self, audio: Any) -> None:
|
||||
"""Write audio samples to the sink."""
|
||||
...
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the audio file."""
|
||||
...
|
||||
@@ -0,0 +1,83 @@
|
||||
"""ConversionRequest — normalized input for a conversion job.
|
||||
|
||||
This is NOT a WebUI Job and NOT a PyQt ConversionThread state.
|
||||
It describes the TASK, not the UI.
|
||||
|
||||
UI adapters are responsible for converting their respective state
|
||||
into a ConversionRequest before calling ConversionService.run().
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionRequest:
|
||||
"""Normalized request for a conversion job.
|
||||
|
||||
Only contains fields that describe the conversion task itself.
|
||||
UI-only fields (display, logging, user prompts) stay in adapters.
|
||||
"""
|
||||
|
||||
# --- Source ---
|
||||
source_path: Optional[Path] = None
|
||||
direct_text: Optional[str] = None
|
||||
original_filename: str = ""
|
||||
|
||||
# --- TTS Settings ---
|
||||
language: str = "a"
|
||||
tts_provider: str = "kokoro"
|
||||
voice: str = "M1"
|
||||
voice_profile: Optional[str] = None
|
||||
speed: float = 1.0
|
||||
use_gpu: bool = True
|
||||
supertonic_total_steps: int = 5
|
||||
|
||||
# --- Output Format ---
|
||||
output_format: str = "wav"
|
||||
subtitle_mode: str = "Disabled"
|
||||
subtitle_format: str = "srt"
|
||||
max_subtitle_words: int = 5
|
||||
|
||||
# --- Save Options ---
|
||||
save_mode: str = "save_next_to_input"
|
||||
output_folder: Optional[Path] = None
|
||||
save_chapters_separately: bool = False
|
||||
merge_chapters_at_end: bool = True
|
||||
separate_chapters_format: str = "wav"
|
||||
save_as_project: bool = False
|
||||
|
||||
# --- Timing ---
|
||||
silence_between_chapters: float = 2.0
|
||||
chapter_intro_delay: float = 0.0
|
||||
|
||||
# --- Content Processing ---
|
||||
replace_single_newlines: bool = False
|
||||
read_title_intro: bool = False
|
||||
read_closing_outro: bool = True
|
||||
auto_prefix_chapter_titles: bool = True
|
||||
normalize_chapter_opening_caps: bool = False
|
||||
|
||||
# --- Pronunciation / Normalization ---
|
||||
pronunciation_overrides: List[Dict[str, Any]] = field(default_factory=list)
|
||||
manual_overrides: List[Dict[str, Any]] = field(default_factory=list)
|
||||
heteronym_overrides: List[Dict[str, Any]] = field(default_factory=list)
|
||||
normalization_overrides: Optional[Dict[str, Any]] = None
|
||||
|
||||
# --- Chapter/Chunk Configuration ---
|
||||
chapter_overrides: List[Dict[str, Any]] = field(default_factory=list)
|
||||
chunks: List[Dict[str, Any]] = field(default_factory=list)
|
||||
chunk_level: str = "paragraph"
|
||||
speaker_mode: str = "single"
|
||||
speakers: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
# --- Metadata ---
|
||||
metadata_tags: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
# --- Artifacts ---
|
||||
cover_image_path: Optional[Path] = None
|
||||
cover_image_mime: Optional[str] = None
|
||||
generate_epub3: bool = False
|
||||
@@ -0,0 +1,47 @@
|
||||
"""ConversionResult — output of a successful conversion.
|
||||
|
||||
Returned by ConversionService.run() after all synthesis and finalization.
|
||||
UI adapters consume this to update their respective state (Job, signals, etc.).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionResult:
|
||||
"""Output of a successful conversion job."""
|
||||
|
||||
# --- Primary outputs ---
|
||||
audio_path: Optional[Path] = None
|
||||
subtitle_paths: List[Path] = field(default_factory=list)
|
||||
chapter_paths: List[Path] = field(default_factory=list)
|
||||
|
||||
# --- Markers (for metadata/audiobookshelf) ---
|
||||
chapter_markers: List[Dict[str, Any]] = field(default_factory=list)
|
||||
chunk_markers: List[Dict[str, Any]] = field(default_factory=list)
|
||||
|
||||
# --- Metadata ---
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
# --- Artifacts ---
|
||||
artifacts: Dict[str, Path] = field(default_factory=dict)
|
||||
project_root: Optional[Path] = None
|
||||
epub_path: Optional[Path] = None
|
||||
|
||||
# --- Stats ---
|
||||
total_chapters: int = 0
|
||||
total_segments: int = 0
|
||||
total_characters: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionError:
|
||||
"""Error information when conversion fails."""
|
||||
|
||||
message: str
|
||||
details: Optional[str] = None
|
||||
is_cancelled: bool = False
|
||||
Reference in New Issue
Block a user