# Coming from Pipecat (Python) > How Pipecat's Python API maps to pipedog - naming, options objects, enums, async and cancellation, frames, and what is not ported. Source: https://polvallverdu.github.io/pipedog/docs/guides/coming-from-pipecat/ pipedog is designed so that someone who knows Pipecat can read pipedog code without a dictionary. Class names, frame names, event names, module paths and wire formats are the same; what changes is what TypeScript forces to change. This page is the short version of the porting rulebook, `docs/CONVENTIONS.md` in the repository. ## A side-by-side example [#a-side-by-side-example] ```python from pipecat.frames.frames import Frame, TextFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class Shout(FrameProcessor): def __init__(self, *, suffix: str = "!", **kwargs): super().__init__(**kwargs) self._suffix = suffix async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, TextFrame): frame = TextFrame(text=frame.text.upper() + self._suffix) await self.push_frame(frame, direction) shout = Shout(suffix="!!", name="shout") @shout.event_handler("on_error") async def on_error(processor, error): print(error.error) ``` ```ts twoslash import { type ErrorFrame, type Frame, TextFrame } from "@pipedog/core/frames/frames"; import { type FrameDirection, FrameProcessor, type FrameProcessorOptions, } from "@pipedog/core/processors/frame_processor"; export interface ShoutOptions extends FrameProcessorOptions { suffix?: string; } class Shout extends FrameProcessor { protected _suffix: string; constructor(options: ShoutOptions = {}) { super(options); this._suffix = options.suffix ?? "!"; } override async processFrame(frame: Frame, direction: FrameDirection): Promise { await super.processFrame(frame, direction); if (frame instanceof TextFrame) { frame = new TextFrame({ text: frame.text.toUpperCase() + this._suffix }); } await this.pushFrame(frame, direction); } } const shout = new Shout({ suffix: "!!", name: "shout" }); shout.eventHandler("on_error", async (_processor: FrameProcessor, error: ErrorFrame) => { console.log(error.error); }); ``` ## Modules and imports [#modules-and-imports] One Python module is one TypeScript file at the same path, with the same snake_case name: | Python | TypeScript | | --------------------------------------------------------------- | --------------------------------------------------------------------------- | | `from pipecat.processors.frame_processor import FrameProcessor` | `import { FrameProcessor } from "@pipedog/core/processors/frame_processor"` | | `from pipecat.services.deepgram.stt import DeepgramSTTService` | `import { DeepgramSTTService } from "@pipedog/deepgram/stt"` (once ported) | | `pipecat.transports.websocket` | `@pipedog/transport-websocket` | | `pipecat.serializers.twilio` | `@pipedog/serializer-twilio` (base serializers stay in core) | | `__init__.py` that re-exports | `index.ts` | The core import closure (frames, processors base, pipeline, workers, bus, observers, RTVI, base transports and services, turns, utils) is one package, `@pipedog/core`. Each provider, transport, serializer and audio model is its own package, like Pipecat's extras. ## Naming [#naming] | Python | TypeScript | | ------------------------------------------------------------- | ------------------------------------------------------------- | | `ClassName` | `ClassName` (unchanged) | | `def snake_case()`, `snake_case` attribute | `camelCase` | | `_protected` | `_protected` (camelCase, keeps the underscore) or `protected` | | `__private` | `#private` | | `CONSTANT` | `CONSTANT` | | event names `"on_pipeline_started"` | **unchanged strings** | | dict keys and JSON on the wire | **unchanged** (a protocol is not an API) | | acronyms: `LLMService`, `start_llm_usage_metrics`, `tts_text` | `LLMService`, `startLlmUsageMetrics`, `ttsText` | ## Constructors and parameters [#constructors-and-parameters] * **Keyword arguments become one options object**, named `Options` and extending the parent's. The whole object is passed to `super(options)`: that is how `**kwargs` flows. * **Required positional parameters of functions stay positional**: `push_frame(frame, direction=DOWNSTREAM)` is `pushFrame(frame, direction = FrameDirection.DOWNSTREAM)`. * **Keyword-only or optional groups become a trailing options object**: `push_error(msg, exception=e)` is `pushError(msg, { exception: e })`. * **`None` becomes `null`**, and defaults use `??` (never `||`, which swallows `0`, `""` and `false`). * **Pydantic params classes** (`PipelineParams`, `InputParams`) become classes with an options constructor and defaults: `new PipelineParams({ enableMetrics: true })`. * **Dataclasses** become classes with fields and an options constructor. ```python worker = PipelineWorker( pipeline, params=PipelineParams(enable_metrics=True, audio_out_sample_rate=24000), idle_timeout_secs=300, ) ``` ```ts const worker = new PipelineWorker(pipeline, { params: new PipelineParams({ enableMetrics: true, audioOutSampleRate: 24000 }), idleTimeoutSecs: 300, }); ``` ## Time, bytes and units [#time-bytes-and-units] * **Durations and timeouts stay in seconds** (floats), as upstream, so configs and docs carry over: `sleep(0.5)`, `idleTimeoutSecs: 300`, `waitFor(fn, 5)`. * **Clock values stay in nanoseconds** as `number` (`frame.pts`). * **`bytes` is `Uint8Array`**, never `Buffer`. PCM is 16-bit little-endian; read samples with `new Int16Array(u8.buffer, u8.byteOffset, u8.length / 2)` (copy first if `byteOffset` is odd). * `time.time()`, `time.monotonic()`, `time.monotonic_ns()` are `time()`, `monotonic()` and `monotonicNs()` from `@pipedog/asyncio`. ## Enums [#enums] Plain enums become an `as const` object plus a type of the same name, keeping upstream's values. `X.FOO.value` is just `X.FOO`: ```ts export const FrameDirection = { DOWNSTREAM: 1, UPSTREAM: 2 } as const; export type FrameDirection = (typeof FrameDirection)[keyof typeof FrameDirection]; ``` Enums with methods or properties (like `ErrorCategory.is_permanent`) become classes with static instances, a `value` field and getters (`ErrorCategory.AUTHENTICATION.isPermanent`), plus `values()` and `fromValue(v)`. `StrEnum`s are `as const` objects with string values. There is no TypeScript `enum` or `namespace` anywhere: the code is erasable so Node can run it directly. ## Async and cancellation [#async-and-cancellation] Python's asyncio semantics come from `@pipedog/asyncio`. The rules that matter most are in [Interruptions & cancellation](https://polvallverdu.github.io/pipedog/docs/concepts/interruptions-and-cancellation.md). | Python | TypeScript | | ---------------------------------------------------------- | ------------------------------------------------------------------------------------- | | `asyncio.create_task(coro())` | `createTask(() => coro())`; in a processor, `this.createTask(() => this.x(), "name")` | | `await self.cancel_task(task)` | `await this.cancelTask(task)` | | `asyncio.sleep(s)` | `sleep(s)` | | `asyncio.Event / Queue / PriorityQueue / Lock / Semaphore` | same names from `@pipedog/asyncio` | | `asyncio.wait_for(coro(), timeout=t)` | `waitFor(() => coro(), t)` (pass a function, so it can be cancelled) | | `asyncio.gather(a(), b())` | `gather(() => a(), () => b())` | | `asyncio.TimeoutError` / `TimeoutError` | `TimeoutError` from `@pipedog/asyncio` | | `asyncio.CancelledError` | `CancelledError` | | `asyncio.current_task()` | `currentTask()` | | `async with lock:` | `await lock.run(async () => { ... })` | | `async for x in aiter` | `for await (const x of aiter)` | | `async def gen(): yield` | `async function* gen()` | | `loop.run_in_executor(None, fn)` | inline if cheap, otherwise an injectable async function (no threads on Workers) | And three rules: 1. **`except Exception:` does not catch `CancelledError`**, but `catch (e)` catches everything. Every broad `catch` starts with `rethrowIfCancelled(e)`. 2. **Foreign awaits** (fetch, an SDK, a WebSocket) are not cancellation points. Pass `currentSignal()` to APIs that take an `AbortSignal`, and/or wrap the await in `abortable()`. 3. **Never leave a promise floating.** Where upstream fires and forgets a task, use `this.createTask(...)` so it is tracked and cancelled at cleanup. ## Classes and inheritance [#classes-and-inheritance] | Python | TypeScript | | -------------------------------------- | ----------------------------------------------------------------- | | abstract methods | `abstract` methods on an `abstract class` | | `@property` (and setter) | `get x()` (and `set x(v)`) | | `@staticmethod`, `@classmethod` | `static` | | `isinstance(x, (A, B))` | `x instanceof A \|\| x instanceof B` | | `type(x).__name__` | `x.constructor.name` | | `__str__` / `__repr__` | `toString()` | | `@obj.event_handler("on_x")` decorator | `obj.eventHandler("on_x", handler)` or `obj.addEventHandler(...)` | **Multiple inheritance**: a second base that is a mixin becomes a class-factory mixin (`class C(A, M)` is `class C extends M(A)`, and `instanceof M` still works). Two concrete bases in the same hierarchy (such as a frame that is both a system frame and a text frame) extend the first and register the second as a virtual base, so `instanceof` works for both. ## Frames [#frames] * Frame constructors **always** take one options object: `TextFrame("hi")` is `new TextFrame({ text: "hi" })`, even where upstream tests pass positional arguments. * `init=True` dataclass fields come from the options; `init=False` fields are initializers. * Frames that are uninterruptible by default declare `override interruptible = false`. * `broadcast_frame(FrameCls, **kwargs)` is `broadcastFrame(FrameCls, options)`. * Field order follows upstream, and `__post_init__` logic runs in the constructor after `super()`. ## Logging [#logging] `from loguru import logger` becomes `import { logger } from "@pipedog/core/utils/logger"`, with the same level methods (`trace`, `debug`, `info`, `warning`, `error`, `critical`). `logger.exception(msg)` is `logger.error(msg, e)`. ## The development runner [#the-development-runner] Pipecat's `pipecat.runner` is `@pipedog/runner`, with the same entry point: export a `bot(runnerArgs)` function and call `main()`. The flags (`-t webrtc`, `-t twilio`, `--proxy`, `--host`, `--port`, ...) are the same. | Python | TypeScript | | ------------------------------------------------------- | ------------------------------------------------------------------------------------------------ | | `from pipecat.runner.run import main` | `import { main } from "@pipedog/runner/run"` | | `if __name__ == "__main__": main()` | `if (import.meta.main) await main();` | | `async def bot(runner_args: RunnerArguments)` | `export async function bot(runnerArgs: RunnerArguments)` | | `await create_transport(runner_args, transport_params)` | `await createTransport(runnerArgs, transportParams)`, from `@pipedog/runner/utils` | | `parse_telephony_websocket(websocket)` | `parseTelephonyWebsocket(websocket)` | | `from pipecat.runner.run import app; @app.get(...)` | `import { app } from "@pipedog/runner/run"; app.get(...)` | | `uvicorn.run(app)` | `serve(app)` from `@pipedog/runner/serve`, or `createRunnerApp()` and its `fetch` in your server | | `python bot.py -t webrtc` | `bun bot.ts -t webrtc` | The FastAPI app is a fetch handler, so the runner mounts in any framework. Daily, MOQ and the Vonage Video Connector are recognized but fail with a clear message: their transports aren't ported yet. See [Development runner](https://polvallverdu.github.io/pipedog/docs/runner.md) for the full mapping. ## What isn't ported yet [#what-isnt-ported-yet] The core, Pipecat Flows, evals, the CLI and most providers are ported; Daily and the other room transports, the Krisp, Koala and AIC filters, the Redis and pgmq buses and a few providers are not. The [roadmap](https://polvallverdu.github.io/pipedog/docs/roadmap/#whats-not-ported-yet) lists the gaps with their upstream module counts. ## Deprecated APIs are not ported [#deprecated-apis-are-not-ported] Anything upstream marks deprecated (with `@deprecated`, `warn_deprecated`, or "will be removed in 2.0.0") is left out of pipedog, including: * the `PipelineTask` / `PipelineTaskParams` / `PipelineRunner` aliases (use `PipelineWorker`, `PipelineParams` and `WorkerRunner`); * `EndTaskFrame` / `CancelTaskFrame` (use `EndWorkerFrame` / `CancelWorkerFrame`); * the `UninterruptibleFrame` mixin (use `interruptible = false`); * deprecated parameters such as `push_error(fatal=)` and `ErrorFrame.fatal` (check `processor.isUsable`, or pass `forceTreatAsPermanent`); * the deprecated `StartFrame` configuration fields (read `FrameProcessorSetup` in `setup()`). What was dropped from each module is recorded in its `notes` in `porting/manifest.json`. ## Tests [#tests] Upstream tests are ported 1:1 with `bun:test`, keeping their names so the two suites can be diffed: `class TestX(unittest.IsolatedAsyncioTestCase)` is `describe("TestX", ...)`, `async def test_y` is `test("test_y", ...)`, and `run_test(...)` is `runTest(...)` from `@pipedog/core/tests/utils`.