feat: add Typer CLI with daemon mode and wtype keyboard injection
Replace argparse CLI with Typer-based CLI supporting `cohere on/off/status` commands. The daemon runs transcription in the background and types into the focused Wayland window via wtype. Adds wtype to flake.nix and fixes the hatchling build backend. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,120 @@
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
|
||||
from ..daemon import STATE_FILE, is_running, read_state, stop_daemon
|
||||
|
||||
app = typer.Typer(help="Cohere live transcription — speaks into your keyboard.")
|
||||
console = Console()
|
||||
|
||||
|
||||
@app.command()
|
||||
def on(
|
||||
language: str = typer.Option("en", "--lang", "-l", help="Language code"),
|
||||
foreground: bool = typer.Option(False, "--fg", help="Run in foreground (don't daemonize)"),
|
||||
):
|
||||
"""Start transcribing and typing into your focused window."""
|
||||
if is_running():
|
||||
console.print("[yellow]Already running.[/yellow]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
if foreground:
|
||||
from ..daemon import run_daemon
|
||||
console.print("[green]Starting cohere (foreground)...[/green]")
|
||||
run_daemon(language)
|
||||
return
|
||||
|
||||
console.print("[green]Starting cohere daemon...[/green]")
|
||||
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
||||
subprocess.Popen(
|
||||
[sys.executable, "-m", "cohere_transcribe.daemon_main", "--lang", language],
|
||||
start_new_session=True,
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=open(os.path.join(os.path.dirname(STATE_FILE), "daemon.log"), "a"),
|
||||
stderr=subprocess.STDOUT,
|
||||
)
|
||||
|
||||
for _ in range(50):
|
||||
time.sleep(0.1)
|
||||
if is_running():
|
||||
break
|
||||
|
||||
if is_running():
|
||||
console.print("[green]Cohere is on — speak and it types.[/green]")
|
||||
else:
|
||||
console.print("[red]Failed to start daemon. Check ~/.local/state/cohere/daemon.log[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
@app.command()
|
||||
def off():
|
||||
"""Stop transcribing."""
|
||||
if not is_running():
|
||||
console.print("[yellow]Not running.[/yellow]")
|
||||
raise typer.Exit(0)
|
||||
|
||||
if stop_daemon():
|
||||
console.print("[red]Cohere is off.[/red]")
|
||||
else:
|
||||
console.print("[red]Failed to stop daemon.[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
@app.command()
|
||||
def status():
|
||||
"""Show whether cohere is running."""
|
||||
state = read_state()
|
||||
running = is_running()
|
||||
|
||||
if running:
|
||||
started = state.get("started_at", 0)
|
||||
elapsed = time.time() - started
|
||||
minutes = int(elapsed) // 60
|
||||
console.print(f"[green]ON[/green] — running for {minutes}m")
|
||||
else:
|
||||
console.print("[dim]OFF[/dim]")
|
||||
|
||||
|
||||
@app.command()
|
||||
def transcribe(
|
||||
audio_file: str = typer.Argument(None, help="Audio file to transcribe"),
|
||||
mic: int = typer.Option(None, "--mic", "-m", help="Record from mic for N seconds"),
|
||||
stream: bool = typer.Option(False, "--stream", "-s", help="Live streaming mode (prints to terminal)"),
|
||||
language: str = typer.Option("en", "--lang", "-l", help="Language code"),
|
||||
):
|
||||
"""One-shot transcription (file, mic, or stream to terminal)."""
|
||||
from ..model import load_model, transcribe_audio
|
||||
|
||||
if stream:
|
||||
from ..stream import stream_transcribe
|
||||
processor, model = load_model()
|
||||
stream_transcribe(processor, model, language)
|
||||
elif mic is not None:
|
||||
from ..model import record_audio
|
||||
processor, model = load_model()
|
||||
try:
|
||||
audio = record_audio(mic)
|
||||
console.print("Transcribing...")
|
||||
text = transcribe_audio(processor, model, audio, language)
|
||||
console.print(f"\n{text}\n")
|
||||
except OSError as e:
|
||||
console.print(f"[red]Microphone error: {e}[/red]")
|
||||
raise typer.Exit(1)
|
||||
elif audio_file:
|
||||
from transformers.audio_utils import load_audio as load_audio_file
|
||||
from ..model import SAMPLE_RATE
|
||||
processor, model = load_model()
|
||||
audio = load_audio_file(audio_file, sampling_rate=SAMPLE_RATE)
|
||||
text = transcribe_audio(processor, model, audio, language)
|
||||
console.print(f"\n{text}\n")
|
||||
else:
|
||||
console.print("[yellow]Provide an audio file, --mic, or --stream[/yellow]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
def main():
|
||||
app()
|
||||
Reference in New Issue
Block a user