Frame processors

function formatMetrics(metrics: MetricsData[]): string {
  return metrics
    .map((metric) => {
      const fields = Object.entries(metric)
        .map(([key, value]) => `\t${key}=${JSON.stringify(value)}`)
        .join("\n");
      return `${metric.constructor.name}\n${fields}`;
    })
    .join("\n");
}

/** Formats and logs every MetricsFrame. */
export class MetricsFrameLogger extends FrameProcessor {
  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);

    if (frame instanceof MetricsFrame) {
      logger.info(`${frame.name}\n${formatMetrics(frame.data)}`);
    }

    // SUPER IMPORTANT: always push every frame!
    await this.pushFrame(frame, direction);
  }
}

const pipeline = new Pipeline([
  transportInput,
  stt,
  userAggregator,
  llm,
  tts,
  transportOutput,
  assistantAggregator,
  new MetricsFrameLogger(), // our custom processor
]);

export interface MyCustomProcessorOptions extends FrameProcessorOptions {
  greeting?: string;
}

export class MyCustomProcessor extends FrameProcessor {
  protected _greeting: string;

  constructor(options: MyCustomProcessorOptions = {}) {
    super(options); // ✅ Required: forwards name, enableDirectMode, metrics...
    this._greeting = options.greeting ?? "hello";
  }

  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction); // ✅ Required

    if (frame instanceof TranscriptionFrame) {
      console.log(`${this._greeting}, user ${frame.userId} said: ${frame.text}`);
    }

    await this.pushFrame(frame, direction); // ✅ Required: pass the frame through
  }
}

export const FrameDirection = { DOWNSTREAM: 1, UPSTREAM: 2 } as const;

// Push downstream (the default)
await this.pushFrame(frame, FrameDirection.DOWNSTREAM);

// Push upstream, back towards the start of the pipeline
await this.pushFrame(new TextFrame({ text: "hi" }), FrameDirection.UPSTREAM);

// Create two instances, push one each way, linked by `broadcastSiblingId`
await this.broadcastFrame(UserStartedSpeakingFrame);

// Copy a frame you didn't create and push it both ways
await this.broadcastFrameInstance(frame);

export class ConnectedProcessor extends FrameProcessor {
  protected _sampleRate = 0;
  protected _client: Client | null = null;

  override async setup(setup: FrameProcessorSetup): Promise<void> {
    await super.setup(setup); // ✅ Required
    this._sampleRate = setup.audioOutSampleRate;
    this._client = await connect();
  }

  override async cleanup(): Promise<void> {
    await super.cleanup(); // ✅ Required
    await this._client?.close();
  }
}

export class SlowToConnect extends FrameProcessor {
  protected _ready = new Event();

  override async setup(setup: FrameProcessorSetup): Promise<void> {
    await super.setup(setup);
    // Hold every frame until connected, for at most 10 seconds.
    await this.pauseProcessingAllFramesUntil(() => this._ready.wait(), { timeout: 10 });
    this.createTask(async () => {
      await connect();
      this._ready.set();
    }, "connect");
  }
}

try {
  await riskyOperation();
} catch (e) {
  // A recoverable error: report it and keep going.
  await this.pushError("Failed to process audio chunk, skipping", { exception: e });
}

// An error that leaves this processor unable to do any more work.
await this.pushError("Critical operation failed", { forceTreatAsPermanent: true });

const tts = new MyCustomProcessor({ name: "tts" });

// Like Python's `@tts.event_handler("on_error")` decorator.
tts.eventHandler("on_error", async (processor: FrameProcessor, error: ErrorFrame) => {
  console.log(`Error in ${processor}: ${error.error}`);
  if (error.exception) {
    console.log("Exception:", error.exception);
  }
  if (error.processor && !error.processor.isUsable) {
    console.log("This processor can no longer do its job");
  }
});

tts.eventHandler("on_usable_changed", async (processor: FrameProcessor, isUsable: boolean) => {
  if (!isUsable) console.warn(`${processor} is down`);
});

// The same handler on several processors.
async function handleServiceError(processor: FrameProcessor, error: ErrorFrame): Promise<void> {
  logger.error(`Service error in ${processor.name}: ${error.error}`);
}
const stt = new MyCustomProcessor({ name: "stt" });
const llm = new MyCustomProcessor({ name: "llm" });
for (const service of [stt, llm, tts]) {
  service.addEventHandler("on_error", handleServiceError);
}

// Frame hooks run inline: keep them fast.
llm.eventHandler("on_before_push_frame", async (processor: FrameProcessor, frame: Frame) => {
  console.log(`${processor} is about to push: ${frame}`);
});

export class ResponseLogger extends FrameProcessor {
  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);

    if (frame instanceof LLMFullResponseStartFrame) {
      logger.info("LLM response started");
    } else if (frame instanceof LLMFullResponseEndFrame) {
      logger.info("LLM response finished");
    }

    await this.pushFrame(frame, direction);
  }
}

import { expect, test } from "bun:test";
import { type Frame, TextFrame } from "@pipedog/core/frames/frames";
import { type FrameDirection, FrameProcessor } from "@pipedog/core/processors/frame_processor";
import { runTest } from "@pipedog/core/tests/utils";

class UpperCaseProcessor extends FrameProcessor {
  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);
    if (frame instanceof TextFrame) {
      await this.pushFrame(new TextFrame({ text: frame.text.toUpperCase() }), direction);
    } else {
      await this.pushFrame(frame, direction);
    }
  }
}

test("test_upper_case", async () => {
  const [down] = await runTest(new UpperCaseProcessor(), {
    framesToSend: [new TextFrame({ text: "hello" })],
    expectedDownFrames: [TextFrame],
  });
  expect((down[0] as TextFrame).text).toEqual("HELLO");
});