modules/organism_perception.py

"""
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("<h", v)
    w.writeframes(bytes(frames)); w.close()
    raw = buf.getvalue()
    return base64.b64encode(raw).decode("ascii"), len(raw), round(n / rate, 3)


def _has(mod: str) -> 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({}))