Implement initial terminal-independent transcription core

7 files changed, 372 insertions, 0 deletions

diff --git a/src/parley/__init__.py b/src/parley/__init__.py
new file mode 100644
index 0000000..8340b53
--- /dev/null
+++ b/src/parley/__init__.py
@@ -0,0 +1,3 @@
+"""Terminal-independent recording and transcription core for Parley."""
+
+__version__ = "0.1.0.dev0"
diff --git a/src/parley/cli.py b/src/parley/cli.py
new file mode 100644
index 0000000..735d532
--- /dev/null
+++ b/src/parley/cli.py
@@ -0,0 +1,74 @@
+"""Standalone terminal frontend for the shared Parley core."""
+
+import sys
+import tempfile
+import termios
+import tty
+from pathlib import Path
+
+from .config import Config
+from .errors import ParleyError
+from .persistence import save_transcript
+from .recording import Recorder
+from .transcription import Transcriber
+
+
+def wait_for_stop() -> bool:
+    """Wait for Enter/Escape and return True when cancellation was requested."""
+    if not sys.stdin.isatty():
+        return input() == "\x1b"
+    descriptor = sys.stdin.fileno()
+    previous_settings = termios.tcgetattr(descriptor)
+    try:
+        tty.setcbreak(descriptor)
+        while True:
+            key = sys.stdin.read(1)
+            if key in ("\r", "\n", ""):
+                return False
+            if key == "\x1b":
+                return True
+    finally:
+        termios.tcsetattr(descriptor, termios.TCSADRAIN, previous_settings)
+
+
+def run_once(config: Config) -> tuple[str, Path]:
+    config.validate()
+    recorder = Recorder(config)
+    with tempfile.TemporaryDirectory(prefix="parley-") as temporary:
+        workspace = Path(temporary)
+        recorder.start(workspace / "recording.wav")
+        print(
+            "Recording; press Enter to transcribe or Escape to cancel.",
+            file=sys.stderr,
+            flush=True,
+        )
+        try:
+            cancelled = wait_for_stop()
+        except BaseException:
+            recorder.cancel()
+            raise
+        if cancelled:
+            recorder.cancel()
+            raise KeyboardInterrupt
+        wav_path = recorder.stop()
+        transcript = Transcriber(config).transcribe(wav_path, workspace)
+        path = save_transcript(transcript, config.transcript_dir)
+        return transcript, path
+
+
+def main() -> int:
+    try:
+        transcript, path = run_once(Config.from_environment())
+        print(transcript)
+        print(f"Saved to {path}", file=sys.stderr)
+        return 0
+    except KeyboardInterrupt:
+        print("Cancelled.", file=sys.stderr)
+        return 130
+    except (OSError, ParleyError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/parley/config.py b/src/parley/config.py
new file mode 100644
index 0000000..88640b6
--- /dev/null
+++ b/src/parley/config.py
@@ -0,0 +1,71 @@
+"""Development configuration for Parley.
+
+Environment variables are intentionally temporary; GSettings will replace them once
+schemas are introduced.
+"""
+
+from dataclasses import dataclass
+import os
+from pathlib import Path
+import shutil
+
+from .errors import ConfigurationError
+
+
+def _default_data_home() -> Path:
+    value = os.environ.get("XDG_DATA_HOME")
+    return Path(value).expanduser() if value else Path.home() / ".local" / "share"
+
+
+def _prototype_root() -> Path:
+    # The source checkout layout documented in README.md.
+    return Path(__file__).resolve().parents[3] / "transcribe-parakeet"
+
+
+@dataclass(frozen=True, slots=True)
+class Config:
+    ffmpeg: Path
+    transcribe_cli: Path
+    model: Path
+    transcript_dir: Path
+    device: str = "default"
+    language: str = "en"
+    sample_rate: int = 16_000
+
+    @classmethod
+    def from_environment(cls) -> "Config":
+        prototype = _prototype_root()
+        ffmpeg = os.environ.get("PARLEY_FFMPEG") or shutil.which("ffmpeg") or "ffmpeg"
+        return cls(
+            ffmpeg=Path(ffmpeg).expanduser(),
+            transcribe_cli=Path(
+                os.environ.get("PARLEY_TRANSCRIBE_CLI", prototype / "bin/transcribe-cli")
+            ).expanduser(),
+            model=Path(
+                os.environ.get(
+                    "PARLEY_MODEL",
+                    prototype / "models/parakeet-unified-en-0.6b-Q8_0.gguf",
+                )
+            ).expanduser(),
+            transcript_dir=Path(
+                os.environ.get(
+                    "PARLEY_TRANSCRIPT_DIR", _default_data_home() / "parley/transcripts"
+                )
+            ).expanduser(),
+            device=os.environ.get("PARLEY_DEVICE", "default"),
+            language=os.environ.get("PARLEY_LANGUAGE", "en"),
+        )
+
+    def validate(self) -> None:
+        if not self.ffmpeg.is_file():
+            raise ConfigurationError(f"ffmpeg not found: {self.ffmpeg}")
+        if not self.transcribe_cli.is_file():
+            raise ConfigurationError(
+                f"transcription executable not found: {self.transcribe_cli}"
+            )
+        if not os.access(self.transcribe_cli, os.X_OK):
+            raise ConfigurationError(
+                f"transcription executable is not executable: {self.transcribe_cli}"
+            )
+        if not self.model.is_file():
+            raise ConfigurationError(f"model not found: {self.model}")
diff --git a/src/parley/errors.py b/src/parley/errors.py
new file mode 100644
index 0000000..7862463
--- /dev/null
+++ b/src/parley/errors.py
@@ -0,0 +1,17 @@
+"""Errors exposed by the Parley core."""
+
+
+class ParleyError(Exception):
+    """Base class for expected, user-actionable Parley failures."""
+
+
+class ConfigurationError(ParleyError):
+    """A required executable, model, or setting is invalid."""
+
+
+class RecordingError(ParleyError):
+    """Audio capture could not start or finish successfully."""
+
+
+class TranscriptionError(ParleyError):
+    """The transcription runtime failed or returned no result."""
diff --git a/src/parley/persistence.py b/src/parley/persistence.py
new file mode 100644
index 0000000..c5bab79
--- /dev/null
+++ b/src/parley/persistence.py
@@ -0,0 +1,51 @@
+"""Collision-safe, atomic transcript persistence."""
+
+from datetime import datetime
+import os
+from pathlib import Path
+import tempfile
+from typing import Callable
+
+
+Clock = Callable[[], datetime]
+
+
+def save_transcript(
+    text: str, directory: Path, *, clock: Clock | None = None
+) -> Path:
+    """Atomically save UTF-8 text and return its collision-safe path."""
+    directory.mkdir(parents=True, exist_ok=True)
+    now = (clock or (lambda: datetime.now().astimezone()))()
+    timestamp = now.strftime("%Y-%m-%d_%H-%M-%S_%f%z")
+    content = text.rstrip("\n") + "\n"
+
+    temporary_name: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=directory,
+            prefix=".transcript-",
+            suffix=".tmp",
+            delete=False,
+        ) as output:
+            temporary_name = output.name
+            output.write(content)
+            output.flush()
+            os.fsync(output.fileno())
+
+        temporary = Path(temporary_name)
+        counter = 0
+        while True:
+            suffix = f"_{counter}" if counter else ""
+            destination = directory / f"transcript_{timestamp}{suffix}.txt"
+            try:
+                # A hard link publishes the completed file atomically and refuses to
+                # replace a name won by another process.
+                os.link(temporary, destination)
+                return destination
+            except FileExistsError:
+                counter += 1
+    finally:
+        if temporary_name is not None:
+            Path(temporary_name).unlink(missing_ok=True)
diff --git a/src/parley/recording.py b/src/parley/recording.py
new file mode 100644
index 0000000..5ee3ec2
--- /dev/null
+++ b/src/parley/recording.py
@@ -0,0 +1,97 @@
+"""FFmpeg recording lifecycle with no terminal or UI dependencies."""
+
+from pathlib import Path
+import subprocess
+
+from .config import Config
+from .errors import RecordingError
+
+
+class Recorder:
+    """Own one FFmpeg process and produce a valid PCM WAV on stop."""
+
+    def __init__(self, config: Config, *, stop_timeout: float = 5.0) -> None:
+        self.config = config
+        self.stop_timeout = stop_timeout
+        self._process: subprocess.Popen[bytes] | None = None
+        self._output_path: Path | None = None
+
+    @property
+    def is_recording(self) -> bool:
+        return self._process is not None and self._process.poll() is None
+
+    def start(self, output_path: Path) -> None:
+        if self._process is not None:
+            raise RecordingError("a recording is already active")
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        command = [
+            str(self.config.ffmpeg),
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-y",
+            "-f",
+            "pulse",
+            "-i",
+            self.config.device,
+            "-ac",
+            "1",
+            "-ar",
+            str(self.config.sample_rate),
+            "-c:a",
+            "pcm_s16le",
+            str(output_path),
+        ]
+        try:
+            self._process = subprocess.Popen(
+                command,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.PIPE,
+            )
+        except OSError as error:
+            raise RecordingError(f"could not start ffmpeg: {error}") from error
+        self._output_path = output_path
+
+    def stop(self) -> Path:
+        process = self._require_process()
+        output_path = self._output_path
+        assert output_path is not None
+        returncode, stderr = self._finish(process)
+        self._clear()
+        if returncode != 0:
+            detail = stderr.decode(errors="replace").strip()
+            output_path.unlink(missing_ok=True)
+            raise RecordingError(f"recording failed: {detail or 'ffmpeg error'}")
+        if not output_path.is_file():
+            raise RecordingError("recording finished without producing audio")
+        return output_path
+
+    def cancel(self) -> None:
+        if self._process is not None:
+            self._finish(self._process)
+        output_path = self._output_path
+        self._clear()
+        if output_path is not None:
+            output_path.unlink(missing_ok=True)
+
+    def _require_process(self) -> subprocess.Popen[bytes]:
+        if self._process is None:
+            raise RecordingError("no recording is active")
+        return self._process
+
+    def _finish(self, process: subprocess.Popen[bytes]) -> tuple[int, bytes]:
+        try:
+            _, stderr = process.communicate(input=b"q\n", timeout=self.stop_timeout)
+        except subprocess.TimeoutExpired:
+            process.terminate()
+            try:
+                _, stderr = process.communicate(timeout=self.stop_timeout)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                _, stderr = process.communicate()
+        return process.returncode, stderr or b""
+
+    def _clear(self) -> None:
+        self._process = None
+        self._output_path = None
diff --git a/src/parley/transcription.py b/src/parley/transcription.py
new file mode 100644
index 0000000..eea0497
--- /dev/null
+++ b/src/parley/transcription.py
@@ -0,0 +1,59 @@
+"""Invoke transcribe-cli and parse its JSONL batch output."""
+
+import json
+from pathlib import Path
+import subprocess
+
+from .config import Config
+from .errors import TranscriptionError
+
+
+def parse_jsonl(output: str) -> str:
+    """Return the first transcript row, ignoring headers and diagnostic lines."""
+    for line in output.splitlines():
+        try:
+            row = json.loads(line)
+        except (json.JSONDecodeError, TypeError):
+            continue
+        if not isinstance(row, dict) or row.get("type") == "batch_header":
+            continue
+        if row.get("error"):
+            raise TranscriptionError(f"transcription failed: {row['error']}")
+        if "text" in row:
+            return str(row["text"])
+    detail = output.strip()
+    if len(detail) > 500:
+        detail = detail[:500] + "…"
+    raise TranscriptionError(f"no transcript in runtime output: {detail or '<empty>'}")
+
+
+class Transcriber:
+    def __init__(self, config: Config) -> None:
+        self.config = config
+
+    def transcribe(self, wav_path: Path, workspace: Path) -> str:
+        batch_file = workspace / "input.txt"
+        batch_file.write_text(f"{wav_path}\n", encoding="utf-8")
+        command = [
+            str(self.config.transcribe_cli),
+            "--quiet",
+            "--model",
+            str(self.config.model),
+            "--language",
+            self.config.language,
+            "--timestamps",
+            "none",
+            "--stream-chunk-ms",
+            "30",
+            "--batch",
+            str(batch_file),
+            "--batch-jsonl",
+        ]
+        try:
+            result = subprocess.run(command, text=True, capture_output=True)
+        except OSError as error:
+            raise TranscriptionError(f"could not start transcription: {error}") from error
+        if result.returncode != 0:
+            detail = result.stderr.strip() or "transcription runtime failed"
+            raise TranscriptionError(detail)
+        return parse_jsonl(result.stdout)