mtf-assistant/src/server.js

// Import dotenv file if not in production
if (process.env.NODE_ENV !== "production") {
  require("dotenv").config();
}

const TinyTTS = require("tiny-tts");
const tts = new TinyTTS();

const fs = require("fs");
const os = require("os");
const path = require("path");
const wav = require("wav-decoder");
const Speaker = require("speaker");
const { spawn } = require("child_process");

const {redis} = require('./modules/redis');

const config = require("./config");
const transConfig = require("./config/transcribe.config");
const { handlePythonOutput } = require("./scripts/py/pythonOutput");

let speaker = null;
let speakerChannels = null;

function ensureSpeaker(channels) {
  if (speaker && speakerChannels === channels) return;
  speakerChannels = channels;

  if (speaker) {
    try { speaker.end(); } catch {}
  }

  const candidates = [
    { device: "default" },
    { device: "plughw:0,0" },
    { device: "hw:0,0" },
  ];

  let lastErr = null;
  for (const c of candidates) {
    try {
      speaker = new Speaker({
        channels,
        bitDepth: 16,
        sampleRate: 44100,
        device: c.device,
      });
      return;
    } catch (e) {
      lastErr = e;
    }
  }

  throw lastErr || new Error("Failed to open output device");
}

function floatsToPCM16LE(channelData) {
  const nCh = channelData.length;
  const n = channelData[0].length;
  const out = Buffer.alloc(n * nCh * 2);

  for (let i = 0; i < n; i++) {
    for (let ch = 0; ch < nCh; ch++) {
      let s = channelData[ch][i];
      s = Math.max(-1, Math.min(1, s));
      out.writeInt16LE((s * 0x7fff) | 0, (i * nCh + ch) * 2);
    }
  }
  return out;
}

async function playAudio(text) {
  const outFile = path.join(
    os.tmpdir(),
    `tinytts-${Date.now()}-${Math.random().toString(16).slice(2)}.wav`
  );

  try {
    // Force tiny-tts to write the WAV to disk (works across versions)
    const maybeBytes = await tts.speak(text, { output: outFile });

    // Some versions return bytes, some return nothing. Always read the file if needed.
    const wavBytes = Buffer.isBuffer(maybeBytes)
      ? maybeBytes
      : await fs.promises.readFile(outFile);

    if (!Buffer.isBuffer(wavBytes) || wavBytes.length < 44) {
      throw new Error("tiny-tts did not produce a valid WAV buffer");
    }
    if (wavBytes.toString("ascii", 0, 4) !== "RIFF") {
      throw new Error("tiny-tts WAV buffer missing RIFF header");
    }

    const decoded = await wav.decode(wavBytes);
    if (!decoded || decoded.sampleRate !== 44100 || !decoded.channelData?.length) {
      throw new Error(`Unsupported WAV: ${decoded?.sampleRate} Hz`);
    }

    ensureSpeaker(decoded.channelData.length);
    const pcm = floatsToPCM16LE(decoded.channelData);
    speaker.write(pcm);
  } finally {
    fs.promises.unlink(outFile).catch(() => {});
  }
}

const py = spawn("python", ["./src/scripts/py/transcribe.py", JSON.stringify(transConfig)], {
  stdio: ["ignore", "pipe", "pipe"],
});

async function processor(group, consumer, streamKey) {
  while (true) {
    const res = await redis.client.xReadGroup(
      group,
      consumer,
      [{ key: streamKey, id: ">" }],
      { COUNT: 5, BLOCK: 3000 }
    );

    if (!res) continue;

    for (const stream of res) {
      for (const msg of stream.messages) {
        if (msg.message.device_id !== config.DEVICE_ID) {
          continue
        }

        console.log(`[${consumer}] Processing:`, msg);

        switch (msg.message.command) {
          case "greet":
            await playAudio("Hello! I am the Many Tailed Fox.");
            break;
          
          case "remind":
            await playAudio("Reminder, " + String(msg.message.value));
            break;

          default:
            await playAudio("Command, " + String(msg.message.command) + " is undefined.")
        }

        await redis.client.xAck(streamKey, group, msg.id);
      }
    }
  }
}

async function main() {
  // Attach python listeners FIRST to avoid missing logs
  py.stdout.on("data", (data) => {
    handlePythonOutput(data.toString(), async (msg) => {
      if (msg.event === "text" || msg.event === "final") {
        await redis.sendTranscriptEvent("transcripts", config.DEVICE_ID, msg.text);
      }
    });
  });

  py.stderr.on("data", (data) => {
    console.error("[PY ERR]", data.toString());
  });

  py.on("close", (code) => {
    console.log("Python exited with code", code);
  });

  try {
    await redis.connect();
    await redis.ensureGroup("commands", "assistant");
    await playAudio("Assistant, online.");
    
    // Start the command processor
    processor("assistant", `${config.DEVICE_ID}`, "commands").catch(console.error);
    
    console.log("System fully initialized.");
  } catch (err) {
    console.error("Initialization failed:", err);
  }
}

main().catch(console.error);