Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion codec.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ def _is_wake_utterance(text: str) -> bool:
# A-4: canonical skill dispatch (lazy SkillRegistry + safety gate + run_with_hooks).
from codec_dispatch import check_skill, run_skill, load_skills

from codec_overlays import show_overlay, show_recording_overlay, show_processing_overlay, show_toggle_overlay
from codec_overlays import show_overlay, show_recording_overlay, show_recording_stop, show_processing_overlay, show_toggle_overlay
from codec_identity import CODEC_VOICE_PROMPT

# ── SKILLS ───────────────────────────────────────────────────────────────────
Expand Down Expand Up @@ -648,6 +648,7 @@ def do_stop_voice():
try: ovl.terminate()
except Exception as e: log.debug("Overlay process cleanup failed: %s", e)
state["overlay_proc"] = None
show_recording_stop() # hide the Swift recording HUD
if not audio or not os.path.exists(audio): return
# Reject recordings shorter than 0.5s — just button taps, not speech
rec_duration = time.time() - rec_start if rec_start else 0
Expand Down Expand Up @@ -886,6 +887,7 @@ def on_release(key):
try: ovl.terminate()
except Exception: pass
state["overlay_proc"] = None
show_recording_stop() # hide the Swift recording HUD on release
threading.Thread(target=lambda: subprocess.run(
['afplay', '/System/Library/Sounds/Pop.aiff'],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL), daemon=True).start()
Expand Down
60 changes: 60 additions & 0 deletions codec_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,67 @@ def focused_app():
log.debug("Focused app detection failed: %s", e)
return "Unknown"

def _swift_text_input(timeout=120):
"""F16 branded glass input via the Swift overlay. Emits an input_request and
waits for the reply file. Returns the typed text, "" on cancel, or None if
the panel never appeared / errored (→ caller falls back to osascript).
Fast-fallback: if no .ack within 2s, the panel didn't show — bail immediately."""
import json as _json
import secrets as _secrets
import time as _time
rid = _secrets.token_hex(4)
base = os.path.expanduser(f"~/.codec/overlay_input_{rid}")
reply, ack = base + ".json", base + ".ack"
events = os.path.expanduser("~/.codec/overlay_events.jsonl")
for p in (reply, ack):
try:
os.remove(p)
except OSError:
pass
try:
with open(events, "a") as f:
f.write(_json.dumps({"type": "input_request", "id": rid,
"prompt": "What can I do for you?"}) + "\n")
except Exception:
return None
# 1) wait up to 2s for the panel to acknowledge it appeared
ack_deadline = _time.time() + 2.0
while _time.time() < ack_deadline:
if os.path.exists(ack):
try:
os.remove(ack)
except OSError:
pass
break
_time.sleep(0.05)
else:
return None # panel never showed → osascript fallback
# 2) panel is up — wait for the user's reply
deadline = _time.time() + timeout
while _time.time() < deadline:
if os.path.exists(reply):
try:
with open(reply) as f:
data = _json.load(f)
os.remove(reply)
return str(data.get("text", "")).strip()
except Exception:
return None
_time.sleep(0.15)
return "" # timed out waiting for input — treat as cancel


def get_text_dialog():
# Branded Swift glass panel (floats over fullscreen, matches the HUD) with an
# automatic fall-through to the native osascript dialog if it isn't available.
try:
import codec_overlays
if getattr(codec_overlays, "_USE_SWIFT", False) and codec_overlays._swift_alive():
result = _swift_text_input()
if result is not None:
return result
except Exception as e:
log.debug("Swift input panel unavailable, using osascript: %s", e)
try:
r = subprocess.run(["osascript", "-e",
'set t to text returned of (display dialog "CODEC - Enter task:" default answer "" with title "CODEC" buttons {"Cancel","Send"} default button "Send")'],
Expand Down
83 changes: 16 additions & 67 deletions codec_dictate.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,10 @@
CHANNELS = 1
CHUNK_DURATION_MS = 30 # ms per audio chunk

# Shared overlay renderer (Swift HUD + tkinter fallback) — unifies dictate's
# Listening / Transcribing / LIVE pills with the rest of CODEC.
import codec_overlays

# ── STATE ────────────────────────────────────────────────────────────────────
recording = False
audio_frames = []
Expand Down Expand Up @@ -78,37 +82,8 @@ def load_model():
def show_overlay():
global overlay_proc
try:
script = """
import tkinter as tk
root = tk.Tk()
root.overrideredirect(True)
root.attributes('-topmost', True)
root.attributes('-alpha', 0.95)
root.configure(bg='#111111')
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
w, h = 680, 88
x = (sw - w) // 2
y = sh - 120
root.geometry(f'{w}x{h}+{x}+{y}')
c = tk.Canvas(root, bg='#111111', highlightthickness=0, width=w, height=h)
c.pack()
c.create_rectangle(2, 2, w-2, h-2, outline='#E8711A', width=2)
dot = c.create_oval(24, 30, 40, 46, fill='#ff3b3b', outline='')
c.create_text(w//2+10, 28, text='Listening \\u2014 release \\u2318 to transcribe', fill='#ffffff', font=('SF Pro Display', 16, 'bold'))
c.create_text(w//2+10, 58, text='Press F5 for hands-free live typing', fill='#777777', font=('SF Pro Display', 12))
def pulse():
cur = c.itemcget(dot,'fill')
c.itemconfig(dot, fill='#ff3b3b' if cur=='#440000' else '#440000')
root.after(500, pulse)
pulse()
root.mainloop()
"""
overlay_proc = subprocess.Popen(
[sys.executable, "-c", script],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
# Shared Swift HUD: orange mark + "Listening / release ⌘ to send".
overlay_proc = codec_overlays.show_recording_overlay("⌘")
except Exception as e:
print(f"[DICTATE] Overlay error: {e}")

Expand All @@ -117,40 +92,16 @@ def hide_overlay():
if overlay_proc:
try:
overlay_proc.terminate()
overlay_proc = None
except Exception:
pass
overlay_proc = None
codec_overlays.show_recording_stop()

# ── SHOW PROCESSING OVERLAY ───────────────────────────────────────────────────
def show_processing():
try:
script = """
import tkinter as tk
import sys
root = tk.Tk()
root.overrideredirect(True)
root.attributes('-topmost', True)
root.attributes('-alpha', 0.93)
root.configure(bg='#0a0a0a')
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
w, h = 520, 90
x = (sw - w) // 2
y = sh - 130
root.geometry(f'{w}x{h}+{x}+{y}')
c = tk.Canvas(root, bg='#0a0a0a', highlightthickness=0, width=w, height=h)
c.pack()
c.create_rectangle(1,1,w-1,h-1, outline='#00aaff', width=1)
c.create_text(w//2, h//2, text='\u26a1 Transcribing...', fill='#00aaff', font=('Helvetica', 13))
root.after(20000, root.destroy)
root.mainloop()
"""
p = subprocess.Popen(
[sys.executable, "-c", script],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
return p
# Shared Swift HUD: blue "Transcribing\u2026" (auto-hides; or hide_overlay()).
return codec_overlays.show_processing_overlay("Transcribing...", duration=20000)
except Exception:
return None

Expand Down Expand Up @@ -298,13 +249,8 @@ def start_live_dictation():
threading.Thread(target=lambda: subprocess.run(
['afplay', '/System/Library/Sounds/Blow.aiff'],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL), daemon=True).start()
# Show overlay — log stderr to /tmp so we can diagnose if it fails to render
_ov_err = open("/tmp/codec_dictate_overlay.log", "w")
live_overlay = subprocess.Popen(
[sys.executable, "-c", _live_overlay_script()],
stdout=subprocess.DEVNULL, stderr=_ov_err
)
print(f"[DICTATE] Overlay subprocess pid={live_overlay.pid}")
# Show the LIVE indicator via the shared Swift HUD (tkinter fallback if down)
live_overlay = codec_overlays.show_live_overlay()
# Start recording loop in thread
live_thread = threading.Thread(target=_live_record_loop, daemon=True)
live_thread.start()
Expand All @@ -320,7 +266,8 @@ def stop_live_dictation():
threading.Thread(target=lambda: subprocess.run(
['afplay', '/System/Library/Sounds/Funk.aiff'],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL), daemon=True).start()
# Kill overlay — tkinter mainloop sometimes ignores SIGTERM, so SIGKILL it
# Hide the LIVE indicator (Swift) + terminate any tkinter fallback process
codec_overlays.show_recording_stop()
if live_overlay:
try: live_overlay.terminate()
except OSError: pass # ProcessLookupError covered (subclass of OSError)
Expand Down Expand Up @@ -393,6 +340,7 @@ def transcribe_and_type(audio_path):
proc_overlay.terminate()
except Exception:
pass
codec_overlays.hide_overlay() # hide the Swift "Transcribing…" HUD

if not text or is_hallucination(text):
print(f"[DICTATE] No speech or hallucination: {text!r}")
Expand Down Expand Up @@ -453,6 +401,7 @@ def transcribe_and_type(audio_path):
proc_overlay.terminate()
except Exception:
pass
codec_overlays.hide_overlay() # hide the Swift "Transcribing…" HUD
finally:
try:
os.unlink(audio_path)
Expand Down
124 changes: 116 additions & 8 deletions codec_overlays.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,76 @@
"""CODEC Overlays — AppKit-based overlays that float above fullscreen apps.
Falls back to tkinter if PyObjC is not available."""
"""CODEC Overlays — render through the always-running Swift CODECOverlay
NSPanel (floats above fullscreen, glass-blurred, branded) by appending JSON
event lines to ~/.codec/overlay_events.jsonl. Falls back to tkinter when the
Swift renderer isn't running."""
import json
import os
import shutil
import subprocess
import sys
import time

# ── Swift CODECOverlay IPC channel ──────────────────────────────────────────
# Primary render path. Flip _USE_SWIFT=False to force the legacy tkinter path.
_USE_SWIFT = True
_EVENTS = os.path.expanduser("~/.codec/overlay_events.jsonl")
_MARK_DEST = os.path.expanduser("~/.codec/overlay_mark.png")
_alive_cache = {"ts": 0.0, "alive": None}


def _ensure_mark():
"""Stage the CODEC hexagon mark where the Swift app loads it (idempotent)."""
try:
if not os.path.exists(_MARK_DEST):
src = os.path.join(os.path.dirname(os.path.abspath(__file__)), "favicon.png")
if os.path.exists(src):
os.makedirs(os.path.dirname(_MARK_DEST), exist_ok=True)
shutil.copyfile(src, _MARK_DEST)
except Exception:
pass


def _emit(event):
"""Append one overlay event for the Swift renderer. Never raises."""
try:
os.makedirs(os.path.dirname(_EVENTS), exist_ok=True)
with open(_EVENTS, "a") as f:
f.write(json.dumps(event) + "\n")
return True
except Exception:
return False


def _swift_alive():
"""True if the Swift CODECOverlay process is running (result cached 5s)."""
now = time.time()
if _alive_cache["alive"] is not None and now - _alive_cache["ts"] < 5:
return _alive_cache["alive"]
alive = False
try:
r = subprocess.run(["pgrep", "-f", "CODECOverlay"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=1)
alive = (r.returncode == 0)
except Exception:
alive = False
_alive_cache["ts"] = now
_alive_cache["alive"] = alive
return alive


def _play_sound(path):
"""Play a system sound off-thread (never blocks the overlay)."""
import threading
threading.Thread(
target=lambda: subprocess.run(["afplay", path],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL),
daemon=True,
).start()


try:
_ensure_mark()
except Exception:
pass


def _has_appkit():
Expand Down Expand Up @@ -198,18 +266,59 @@ def pulse():


def show_overlay(text, color="#E8711A", duration=2500):
if _USE_SWIFT and _swift_alive():
_emit({"type": "notify", "text": text, "color": color,
"duration": (duration / 1000.0) if duration else 2.5})
return None
if _USE_APPKIT:
return _appkit_overlay(text, color, duration)
return _tk_overlay(text, color, duration)


def show_recording_overlay(key_label="F18"):
if _USE_SWIFT and _swift_alive():
_emit({"type": "recording_start", "title": "Listening",
"subtitle": f"release {key_label} to send"})
return None
if _USE_APPKIT:
return _appkit_overlay(f"\U0001f3a4 Recording — release {key_label} to send", "#E8711A", duration=0)
return _tk_recording(key_label)


def show_recording_stop():
"""Hide a persistent recording/live overlay. Swift only — the tkinter
fallback is a child process the caller terminates directly."""
if _USE_SWIFT and _swift_alive():
_emit({"type": "recording_stop"})


def hide_overlay():
"""Hide whatever overlay is currently showing (Swift)."""
if _USE_SWIFT and _swift_alive():
_emit({"type": "hide"})


def show_live_overlay():
"""CODEC Dictate F5 hands-free live-typing indicator (red, persistent)."""
if _USE_SWIFT and _swift_alive():
_emit({"type": "live"})
return None
return _tk_overlay("LIVE · press F5 to stop", "#ff3b3b", 0)


def show_refining_overlay():
"""CODEC Dictate draft-refinement indicator (blue, persistent)."""
if _USE_SWIFT and _swift_alive():
_emit({"type": "refining"})
return None
return _tk_overlay("Refining…", "#00aaff", 0)


def show_processing_overlay(text="Transcribing...", duration=4000):
if _USE_SWIFT and _swift_alive():
_emit({"type": "transcribing", "text": text,
"duration": (duration / 1000.0) if duration else 0})
return None
if _USE_APPKIT:
return _appkit_overlay(f"\u26a1 {text}", "#00aaff", duration)
# tkinter fallback
Expand Down Expand Up @@ -239,15 +348,14 @@ def show_processing_overlay(text="Transcribing...", duration=4000):


def show_toggle_overlay(is_on, shortcuts=""):
snd = '/System/Library/Sounds/Blow.aiff' if is_on else '/System/Library/Sounds/Funk.aiff'
_play_sound(snd)
if _USE_SWIFT and _swift_alive():
_emit({"type": "toggle_on", "shortcuts": shortcuts} if is_on else {"type": "toggle_off"})
return None
color = '#E8711A' if is_on else '#ff3333'
label = 'C O D E C' if is_on else 'S I G N I N G O U T'
dur = 3000 if is_on else 1500
import threading
snd = '/System/Library/Sounds/Blow.aiff' if is_on else '/System/Library/Sounds/Funk.aiff'
threading.Thread(
target=lambda: subprocess.run(['afplay', snd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL),
daemon=True
).start()
if _USE_APPKIT:
return _appkit_overlay(label, color, dur, font_size=18, bold=True, subtitle=shortcuts)
# tkinter fallback
Expand Down
Loading