Kokoro working, updated interface

.env-defaults

@@ -0,0 +1,2 @@
+HF_TOKEN=
+NLTK_DATA=

README.md

@@ -33,7 +33,15 @@ git submodule update --init --recursive
 
 # Requirements
 
-Recent Python (>=3.11), pip (>=22.3), some sort of virtula environment recommended (such as the builtin venv package), and a Rust compatible platform (avoid embedded systems which do not have pre-built binaries for the compiled requirements, e.g., Android's Termux try to build everything from sources and fails miserably when doing so).
+- `ffmpeg` available (your distribution's version should be ok)
+- Recent Python (>=3.11), pip (>=22.3), and some sort of virtual environment recommended (such as the builtin venv package).
+- a Rust compatible platform. Avoid embedded systems which do not have pre-built
+  binaries for the compiled requirements (e.g., Android's Termux try to build
+  everything from sources and fails miserably when doing so).
+- For GPU use, make sure you are installing compatible versions of
+  [`pytorch`](https://pytorch.org/get-started/locally/) and adjust your backends
+  installation method, for instance, run `pip install --force-reinstall` with the correct args
+  inside the relevant virtual environments.
 
 # Installation
 

src/epub_tts/__init__.py

@@ -15,5 +15,20 @@ except PackageNotFoundError:
 
 from typing import Union
 from pathlib import Path
+import os
+import pprint
+
+def log (message):
+    pp = pprint.PrettyPrinter(
+        indent=4, 
+        width=os.get_terminal_size().columns, 
+        compact=False, 
+        depth=None
+        )
+    if isinstance(message, (dict, list, tuple, set)):
+        formatted = pp.pformat(message)
+        print(formatted.replace("', '", "',\n'"))
+    else:
+        pp.pprint(message)
 
 PathLike = Union[str, Path]

src/epub_tts/__main__.py

@@ -28,7 +28,11 @@ import sys
 from pathlib import Path
 from argparse import ArgumentParser
 
+from dotenv import load_dotenv
 from . import tts_aedocw,  tts_generic
+from . import log, PathLike
+
+load_dotenv()  # load environment variables from .env file if present
 
 backend = {
     # keys are the backend names, values are the corresponding TTS classes
@@ -55,8 +59,12 @@ def main(args=None):
     )
     p.add_argument("-i", "--input", help="Input EPUB file", required=True)
     p.add_argument("-o", "--output", help="Output audio file", required=True)
-    p.add_argument("-l", "--language", help="Language to use for TTS")
+    p.add_argument("-l", "--language", help="Language to use for TTS (see your backend's documentation for available voices)")
-    p.add_argument("-v", "--voice", help="Voice to use for TTS")
+    p.add_argument("-v", "--voice", help="Voice to use for TTS (see your backend's documentation for available voices)")
+    p.add_argument("--speed", type=float, default=1.0, help="Playback speed multiplier (default: 1.0) (not all backends support this)")
+    p.add_argument("--short-pause", type=int, default=None, help="Short pause duration in milliseconds between phrases or sentences (not all backends support this)")
+    p.add_argument("--long-pause", type=int, default=None, help="Long pause duration in milliseconds between sections or paragraphs (not all backends support this)")
+    p.add_argument("-c", "--cover", help="Path to cover image to embed or use for output metadata")
     p.add_argument("-r", "--replace", action="append", nargs=2, help="Replace text in the intermediate output. Specify pairs of old_text new_text. Can be used multiple times.")
     p.add_argument(
         "-b",
@@ -68,18 +76,19 @@ def main(args=None):
     args = p.parse_args(args or sys.argv[1:])
     setattr(args, 'replace_map', 
             {old: new for old, new in args.replace} if args.replace else None)
-    print(
+    log(args)
-        f"Converting {args.input} to {args.output} "
+    # log(
-        f"using {args.backend} backend "
+    #     f"Converting {args.input} to {args.output} "
-        "and args: " + ", ".join(
+    #     f"using {args.backend} backend "
-            f"{k}={v}" for k, v in vars(args).items() 
+    #     "and args: " + ", ".join(
-            if k not in ("input", "output", "backend") and v is not None)
+    #         f"{k}={v}" for k, v in vars(args).items() 
-    )
+    #         if k not in ("input", "output", "backend") and v is not None)
+    # )
     
-    #print(args.replace_map)
+    #log(args.replace_map)
 
    
-    backend[args.backend]().run(
+    backend[args.backend](**vars(args)).run(
         args.input, args.output, 
         **vars(args),
         )

src/epub_tts/tts_aedocw.py

@@ -14,7 +14,7 @@ corresponding console script is not available in the current environment.
 """
 
 from .tts_generic import GenericTTSBackend
-from . import PathLike
+from . import log, PathLike
 
 class AedocwBackend(GenericTTSBackend):
     """Generic aedocw backend wrapper.
@@ -33,10 +33,17 @@ class AedocwBackend(GenericTTSBackend):
             "speaker": "--speaker",
             "voice": "--speaker",
             "language": "--language",
+            "cover": "--cover",
         }
     INTERMEDIATE_TXT = True
     INTERMEDIATE_CALL = GenericTTSBackend._replace_map
-    
+
+    def gen_input_flag(self, input_path) -> list[str]:
+            """Return backend flag for input file or directory."""
+            p = self._normalize_path(input_path)
+            return [str(p)]  # epub2tts-edge expects a single path, no flag
+            
+    gen_output_flag = None
                                               
     
 
@@ -48,15 +55,10 @@ class AedocwEpub2TTS(AedocwBackend):
 class AedocwEpub2TTSEdge(AedocwBackend):
     """Backend configured for github.com/aedocw/epub2tts-edge."""
     DEFAULT_SPEAKER = "en-US-AndrewNeural"
-    CMD = "epub2tts-edge"
+    CMD = ["-c", "from epub2tts_edge import main;main()"]
     REPO = "epub2tts-edge"
 
-    def gen_input_flag(self, input_path) -> list[str]:
+   
-        """Return backend flag for input file or directory."""
-        p = self._normalize_path(input_path)
-        return [str(p)]  # epub2tts-edge expects a single path, no flag
-        
-    gen_output_flag = None
         
 
 
@@ -73,10 +75,39 @@ class AedocwChatterbox(AedocwBackend):
 
 class AedocwKokoro(AedocwBackend):
     """Backend configured for github.com/aedocw/epub2tts-kokoro."""
-    DEFAULT_SPEAKER = "af_heart"
+    #DEFAULT_SPEAKER = "af_heart"
+    DEFAULT_SPEAKER = "am_liam"
+    CMD = ["-c", "from epub2tts_kokoro import main;main()"]
+    REPO = "epub2tts-kokoro"
+    FLAGS = {
+        "--speaker": "voice", # str, default="af_heart"
+        "--cover": "cover", # str, default=None
+        "--paragraphpause": "long_pause", # int, default=600 (ms)
+        "--speed": "speed", # float, default=1.3
+    }
+    DEFAULT_FLAGS = {
+        "voice": "am_liam",
+        "cover": None,
+        "long_pause": 600,
+        "speed": 1.3,
+    }
 
 
 
+    def get_speakers(self) -> list[str]:
+        """Return list of available speakers."""
+        return ["af_heart", "af_joy", "af_sad", "af_angry", "af_fear", "af_surprise"]
+    
+    def gen_speaker_samples(
+            self, 
+            samples: list =None,
+            output_path: PathLike=None,
+        ) -> list[str]:
+        """Generate sample speakers audio in output_path."""
+        result = self._build_command("gen_samples.py", *samples, output_path=output_path)
+        return [str(self._normalize_path(s)) for s in samples]
+
+
 class Epub2TTS(AedocwBackend):
     """Backward-compatible chatterbox backend wrapper."""
 

src/epub_tts/tts_generic.py

@@ -28,13 +28,14 @@ Classes:
             run(input_source: PathLike, output_dest: PathLike) -> subprocess.CompletedProcess: Execute backend command.
 
 """
+import os
 import re
 import subprocess
 from pathlib import Path
 from typing import Optional, Callable
 
-from .vendor_utils import build_run_command
+from .vendor_utils import build_run_command, ensure_venv
-from . import PathLike
+from . import log, PathLike
 
 class GenericTTSBackend:
     """Generic TTS backend for handling text-to-speech conversion.
@@ -47,6 +48,7 @@ class GenericTTSBackend:
     REPO: Optional[str] = None
     CMD: Optional[str] = None
     FLAGS: Optional[dict] = {}
+    DEFAULT_FLAGS: Optional[dict] = {}
     INTERMEDIATE_TXT: Optional[bool] = False
     INTERMEDIATE_CALL: Optional[Callable] = None
     DEFAULT_SPEAKER: Optional[str] = None
@@ -62,10 +64,20 @@ class GenericTTSBackend:
     
     INPUT_FLAG = {True: "--input-dir", False: "--input-file"}
     OUTPUT_FLAG = {True: "--output-dir", False: "--output-file"}
+    ENV = None  # subclasses may override this to set environment variables for the backend
 
-    def __init__(self,**kwargs):
+    def __init__(self, **kwargs):
+        log(kwargs)
         for k,v in kwargs.items():
             setattr(self, k, v)
+        if hasattr(self, 'CWD') and self.CWD is not None:
+            self.CWD = Path(self.CWD).resolve()
+        else:
+            self.CWD = self._normalize_path(self.output).resolve() if hasattr(self, 'output') else Path.cwd()
+            if not self.CWD.is_dir():
+                self.CWD = self.CWD.parent.resolve()
+        self.ENV = os.environ.copy()
+        #log(f"SELF:{self.__dict__}")
 
 
     def _normalize_path(self, value: PathLike) -> Path:
@@ -85,18 +97,23 @@ class GenericTTSBackend:
         return [self.OUTPUT_FLAG[p.is_dir()], str(p)]
 
     def _build_command(self, input_source: PathLike, output_dest: PathLike) -> list[str]:
-        if self.REPO is None or self.CMD is None:
+        if not self.CMD and not self.REPO:
+            raise ValueError("Either CMD or REPO must be specified for the backend.")
+        if self.CMD and not self.REPO:
             command = [self.CMD]
+        elif self.REPO and not self.CMD:
+            command = [str(ensure_venv(self.REPO))]
         else:
-            command = build_run_command(self.REPO, CMD=self.CMD)
+            command = [str(ensure_venv(self.REPO))]
+            command.extend(["-m",self.CMD] if isinstance(self.CMD, str) 
+                           else self.CMD)
         command.extend(self.gen_input_flag(input_source))
         if self.gen_output_flag:
             command.extend(self.gen_output_flag(output_dest))
         for flag_name, var_name in self.FLAGS.items():
-            if getattr(self, flag_name, None) is not None:
+            if getattr(self, var_name, self.DEFAULT_FLAGS[var_name]) is not None:
-                command.extend([
+                command.extend([flag_name, str(getattr(self, var_name))
-                    f"--{flag_name.replace('_', '-')}", 
+                    ])
-                    getattr(self, var_name)])
         return command
     
     def run(self,
@@ -106,12 +123,15 @@ class GenericTTSBackend:
             ) -> subprocess.CompletedProcess:
 
         command = self._build_command(input_source, output_dest)
-        print(command)
+        log(command)
-        print(f"Running command: {' '.join(command)}")
+
-        completed = subprocess.run(
+        # PREPROCESSING STEP: If INTERMEDIATE_TXT is True, run the command to generate intermediate text first
-            self._build_command(input_source, output_dest), 
+        if not input_source.endswith(".txt"):
-            cwd=str(self._normalize_path(input_source).parent), 
+            completed = subprocess.run(
-            check=True)
+                self._build_command(input_source, output_dest), 
+                cwd=str(self.CWD), 
+                env=self.ENV,
+                check=True)
         
         if self.INTERMEDIATE_TXT:
             txt_file = self._normalize_path(input_source).with_suffix(".txt")
@@ -124,11 +144,11 @@ class GenericTTSBackend:
                     txt_file = txt_file.with_stem(txt_file.stem + "_replaced")
                 completed = subprocess.run(
                     self._build_command(txt_file, output_dest),
-                    cwd=str(self._normalize_path(input_source).parent),
+                    cwd=str(self.CWD),
+                    env=self.ENV,
                     check=True
                 )
 
-
         return completed
 
        

src/epub_tts/vendor_utils.py

@@ -10,7 +10,7 @@ import sys
 import venv
 from pathlib import Path
 from typing import Optional
-
+from . import log, PathLike
 
 def get_vendor_root() -> Path:
     return Path(__file__).resolve().parents[1] / "vendor"
@@ -57,24 +57,17 @@ def ensure_venv(repo_name: str) -> Path:
 
 
 def build_run_command(
-    repo_name: str,
+        repo_name: str,
-    *,
+        # backend_cmd: str = None,
-    MOD: Optional[str] = None,
+    ) -> str:
-    CMD: Optional[str] = None,
+    return str(ensure_venv(repo_name))
-) -> list[str]:
+    # if not backend_cmd:
-    if not MOD and not CMD:
+    #     return str(ensure_venv(repo_name))
-        raise ValueError("Either MOD or CMD must be provided.")
+    # ensure_venv(repo_name)
-    if MOD:
+    # venv_path = get_venv_path(repo_name)
-        python_executable = ensure_venv(repo_name)
+    # return (
-        return [str(python_executable), "-m", MOD]
+    #     str( venv_path / "Scripts" / f"{backend_cmd}.exe")
-    if CMD:
+    #     if sys.platform == "win32"
-        ensure_venv(repo_name)
+    #     else str(venv_path / "bin" / backend_cmd)
-        venv_path = get_venv_path(repo_name)
+    #     )
-        return [str( venv_path / "Scripts" / f"{CMD}.exe")
-            if sys.platform == "win32"
-            else str(venv_path / "bin" / CMD)]
-    # else:
-    python_executable = ensure_venv(repo_name)
-    return [str(python_executable), "-m", MOD]
-