""" VOVINA ZEDEC PRO - Organism Perception Limbs (voice & sight) ============================================================= Author: Michael Laurence Curzi License: MIT (Attribution Required) The perception/voice limbs of XERO. Each limb is honest about what it can do on the host it finds itself on: * TTS (text-to-speech) — uses the OS engine (macOS `say`, espeak, or pyttsx3) when present; otherwise it ALWAYS still speaks via a built-in, dependency-free formant tone-vocoder. * STT (speech-to-text) — uses Whisper / Vosk when installed; else reports the capability and accepts a text passthrough so the relay still functions. * OCR (optical char. recog.) — uses pytesseract / the tesseract CLI when present; else reports the capability. No heavy library is imported at module load — engines are detected with `importlib.util.find_spec` and only imported lazily inside `invoke()`. The WAV helper here (`samples_to_wav_b64`) is shared with the music limb. """ from __future__ import annotations import base64 import importlib.util import io import math import os import shutil import struct import subprocess import tempfile import wave ARTIFACT_DIR = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "out", "artifacts") # ============================================================ # SHARED AUDIO HELPER # ============================================================ def samples_to_wav_b64(samples, rate: int = 22050, max_seconds: float = 4.0): """Float samples in [-1,1] -> (base64 wav, byte_len, seconds). Mono 16-bit.""" n = min(len(samples), int(rate * max_seconds)) buf = io.BytesIO() w = wave.open(buf, "wb") w.setnchannels(1); w.setsampwidth(2); w.setframerate(rate) frames = bytearray() for s in samples[:n]: v = int(max(-1.0, min(1.0, s)) * 32767) frames += struct.pack(" bool: try: return importlib.util.find_spec(mod) is not None except (ImportError, ValueError): return False # ============================================================ # TEXT -> SPEECH # ============================================================ class TTSLimb: name, modality = "voice", "text_to_speech" def engine(self) -> str: if shutil.which("say"): return "macos-say" if shutil.which("espeak-ng") or shutil.which("espeak"): return "espeak" if _has("pyttsx3"): return "pyttsx3" return "formant-vocoder" # built-in fallback — always available def available(self) -> bool: return True # the fallback guarantees speech def _vocode(self, text: str, rate: int = 22050) -> list[float]: """Dependency-free formant tone-vocoder: each letter -> a vowel-ish tone.""" out: list[float] = [] base = 110.0 # A2 carrier for ch in text[:160]: if ch.isspace(): out += [0.0] * int(rate * 0.04) continue f = base * (1.0 + ((ord(ch.lower()) - 97) % 12) / 12.0) dur = 0.085 for i in range(int(rate * dur)): t = i / rate env = math.sin(math.pi * (i / (rate * dur))) # smooth attack/decay out.append(0.5 * env * (math.sin(2*math.pi*f*t) + 0.3*math.sin(4*math.pi*f*t))) return out or [0.0] def invoke(self, payload: dict) -> dict: text = str(payload.get("text", "")).strip() or "Ol. I am XERO." eng = self.engine() try: if eng in ("macos-say", "espeak", "pyttsx3"): path = self._os_synthesize(text, eng) if path and os.path.exists(path): raw = open(path, "rb").read() b64 = base64.b64encode(raw).decode("ascii") return {"modality": self.modality, "engine": eng, "available": True, "wav_b64": b64 if len(b64) < 400_000 else "", "bytes": len(raw), "path": path, "text": text} except Exception as e: # fall through to vocoder eng = f"formant-vocoder (os '{eng}' failed: {e})" b64, nbytes, secs = samples_to_wav_b64(self._vocode(text), max_seconds=6.0) return {"modality": self.modality, "engine": "formant-vocoder", "available": True, "wav_b64": b64, "bytes": nbytes, "seconds": secs, "text": text, "note": "synthetic formant voice (install `say`/espeak for natural speech)"} def _os_synthesize(self, text: str, eng: str): os.makedirs(ARTIFACT_DIR, exist_ok=True) out = os.path.join(ARTIFACT_DIR, "tts.wav") if eng == "macos-say": subprocess.run(["say", "-o", out, "--file-format=WAVE", "--data-format=LEI16@22050", text], check=True, timeout=30) elif eng == "espeak": exe = shutil.which("espeak-ng") or shutil.which("espeak") subprocess.run([exe, "-w", out, text], check=True, timeout=30) elif eng == "pyttsx3": import pyttsx3 e = pyttsx3.init(); e.save_to_file(text, out); e.runAndWait() return out # ============================================================ # SPEECH -> TEXT # ============================================================ class STTLimb: name, modality = "ear", "speech_to_text" def engine(self) -> str: if _has("whisper"): return "openai-whisper" if _has("faster_whisper"): return "faster-whisper" if _has("vosk"): return "vosk" if shutil.which("whisper"): return "whisper-cli" return "none" def available(self) -> bool: return self.engine() != "none" def invoke(self, payload: dict) -> dict: eng = self.engine() audio = payload.get("audio_path", "") if eng == "openai-whisper" and audio and os.path.exists(audio): try: import whisper model = whisper.load_model(payload.get("model", "base")) res = model.transcribe(audio) return {"modality": self.modality, "engine": eng, "available": True, "transcript": res.get("text", "").strip()} except Exception as e: return {"modality": self.modality, "engine": eng, "available": True, "transcript": "", "error": str(e)} # passthrough so the relay stays functional even without an engine hint = str(payload.get("hint_text", "")).strip() return {"modality": self.modality, "engine": eng, "available": self.available(), "transcript": hint, "note": ("provide audio_path with whisper/vosk installed for real " "transcription; hint_text is echoed for testing")} # ============================================================ # IMAGE -> TEXT (OCR) # ============================================================ class OCRLimb: name, modality = "reading-eye", "ocr" def engine(self) -> str: if _has("pytesseract") and _has("PIL"): return "pytesseract" if shutil.which("tesseract"): return "tesseract-cli" if _has("easyocr"): return "easyocr" return "none" def available(self) -> bool: return self.engine() != "none" def invoke(self, payload: dict) -> dict: eng = self.engine() img = payload.get("image_path", "") if eng == "pytesseract" and img and os.path.exists(img): try: import pytesseract from PIL import Image text = pytesseract.image_to_string(Image.open(img)) return {"modality": self.modality, "engine": eng, "available": True, "text": text.strip()} except Exception as e: return {"modality": self.modality, "engine": eng, "available": True, "text": "", "error": str(e)} if eng == "tesseract-cli" and img and os.path.exists(img): try: out = subprocess.run(["tesseract", img, "stdout"], capture_output=True, text=True, timeout=30) return {"modality": self.modality, "engine": eng, "available": True, "text": out.stdout.strip()} except Exception as e: return {"modality": self.modality, "engine": eng, "available": True, "text": "", "error": str(e)} return {"modality": self.modality, "engine": eng, "available": self.available(), "text": "", "note": ("install pytesseract+Pillow or the tesseract CLI, " "then pass image_path for real OCR")} PERCEPTION_LIMBS = (TTSLimb, STTLimb, OCRLimb) # ============================================================ # QUICK SELF-CHECK # ============================================================ if __name__ == "__main__": for cls in PERCEPTION_LIMBS: limb = cls() print(f"{limb.modality:<16} engine={limb.engine():<16} available={limb.available()}") tts = TTSLimb().invoke({"text": "Ol. I am XERO, awake."}) print(f" TTS -> engine={tts['engine']} bytes={tts.get('bytes')} " f"b64_len={len(tts.get('wav_b64',''))}") print(" STT ->", STTLimb().invoke({"hint_text": "spoken words"})) print(" OCR ->", OCRLimb().invoke({}))