Media¶

The rtpengine namespace controls media anchoring and injection (announcements, DTMF, gating, subscriptions) via the RTPEngine / siphon-rtp NG control protocol. The qos namespace turns an SDP offer/answer pair into the media_components structure that diameter.rx_aar and sbi.create_session consume.

from siphon import rtpengine

@b2bua.on_invite
async def anchor(call):
    await rtpengine.play_media(call, file="/prompts/welcome.wav")

`rtpengine` namespace¶

Mock RTPEngine namespace — records media operations for assertions.

Example::

from siphon import rtpengine
# After running handler:
assert rtpengine.operations == [("offer", "srtp_to_rtp")]

Media-injection operations (play_media, stop_media, play_dtmf, silence_media, unsilence_media, block_media, unblock_media, echo) are also captured in operations as (name, detail) tuples so downstream apps can unit-test MMTEL announcement flows without a live rtpengine. Full parameter dicts are available on media_calls.

Every media verb's target accepts three forms (like the runtime's resolve_call_from_tag): a SIP object (Request/Reply/Call), a (call_id, from_tag) pair, or a bare call_id string — so an @rtpengine.on_dtmf handler can drive media from the call_id / from_tag it was handed. The resolved call_id / from_tag are recorded on each media_calls entry.

Valid profiles: "srtp_to_rtp", "ws_to_rtp", "wss_to_rtp", "rtp_passthrough".

operations `instance-attribute` ¶

operations: list[tuple[str, Optional[str]]] = []

List of (operation, profile_or_detail) tuples recorded.

media_calls `instance-attribute` ¶

media_calls: list[dict[str, Any]] = []

Full parameter dicts for each media-injection call.

active_sessions `property` ¶

active_sessions: int

Number of active media sessions (mock: count of offer - delete).

instance_count `property` ¶

instance_count: int

Number of configured RTPEngine instances (mock: always 1).

offer `async` ¶

offer(request: Any, profile: Optional[str] = None) -> bool

Send offer command to RTPEngine.

Extracts SDP from message body, sends to engine, replaces body with rewritten SDP.

Parameters:

Name	Type	Description	Default
`request`	`Any`	Request or Call object with SDP body.	required
`profile`	`Optional[str]`	RTP profile name. Defaults to `"rtp_passthrough"`.	`None`

Returns:

Type	Description
`bool`	`True` on success.

answer `async` ¶

answer(
    reply: Any,
    profile: Optional[str] = None,
    call: Any = None,
) -> bool

Send answer command to RTPEngine.

Profile precedence (matches the real implementation):

Explicit profile= argument (script override).
Profile recorded by the matching offer (looked up by A-leg Call-ID). Lets @b2bua.on_answer / @b2bua.on_early_media call rtpengine.answer(reply) with no profile= and still get the directional flags from the offer-side profile.
"rtp_passthrough" when no offer was ever recorded.

Parameters:

Name	Type	Description	Default
`reply`	`Any`	Reply or Call object with SDP body.	required
`profile`	`Optional[str]`	Optional explicit RTP profile name. When omitted, the profile recorded by the matching offer is used.	`None`
`call`	`Any`	Optional Call object — when provided, Call-ID and From-tag are taken from this object (matching the earlier `offer`), while To-tag and SDP body still come from `reply`.	`None`

Returns:

Type	Description
`bool`	`True` on success.

answer_local `async` ¶

answer_local(
    call: Any,
    profile: Optional[str] = None,
    auto_reject: bool = True,
) -> Optional[str]

Single-leg UAS answer — synthesise an RFC 3264 answer for the caller's own offer, with the media engine as the far side (IVR / echo / announcement server).

Unlike :meth:answer, this takes the offer (INVITE), not a peer's reply: there is no far leg, so the engine picks one encodable codec from the offer and returns the answer SDP for the script to put in its own 2xx.

Profile precedence matches :meth:answer (explicit profile= → profile recorded by a matching offer → "rtp_passthrough").

When the offer has no encodable codec (primed in tests via :meth:set_answer_local_no_codec), the engine cannot answer:

with auto_reject=True (default) and a Call target, a deferred 488 Not Acceptable Here is recorded on the call (call.reject(488, "Not Acceptable Here")) and the coroutine resolves to None;
with auto_reject=False (or a non-Call target) it raises ValueError instead.

Native siphon-rtp backend only.

Parameters:

Name	Type	Description	Default
`call`	`Any`	A `Call` (B2BUA) — or `Request` — carrying the INVITE offer whose Call-ID / From-tag drive the single-leg answer.	required
`profile`	`Optional[str]`	Optional explicit RTP profile name. When omitted, the profile recorded by a matching offer is used.	`None`
`auto_reject`	`bool`	When `True` (default) and `call` is a `Call`, a no-encodable-codec result records a deferred 488 and returns `None`. When `False` it raises `ValueError`.	`True`

Returns:

Type	Description
`Optional[str]`	The answer SDP as `str` on success, or `None` when the offer had
`Optional[str]`	no encodable codec and it was auto-rejected with a 488.

Example::

@b2bua.on_invite
async def on_invite(call):
    sdp = await rtpengine.answer_local(call, profile="ivr")
    if sdp is not None:
        call.answer(200, "OK", body=sdp, content_type="application/sdp")
        await rtpengine.play_media(call, file="/prompts/welcome.wav")

delete `async` ¶

delete(request: Any) -> bool

Send delete command to tear down media session.

Parameters:

Name	Type	Description	Default
`request`	`Any`	Request or Call object (uses Call-ID + From-tag).	required

Returns:

Type	Description
`bool`	`True` on success.

ping `async` ¶

ping() -> bool

Health check: ping RTPEngine instance(s).

Returns:

Type	Description
`bool`	`True` if healthy.

play_media `async` ¶

play_media(
    target: Any,
    file: Optional[str] = None,
    blob: Optional[bytes] = None,
    db_id: Optional[int] = None,
    repeat: Optional[int] = None,
    start_ms: Optional[int] = None,
    duration_ms: Optional[int] = None,
    to_tag: Optional[str] = None,
    wait: bool = True,
) -> Optional[int]

Inject an audio prompt into the call.

Exactly one of file/blob/db_id must be supplied. Per rtpengine semantics, from-tag (derived from target) selects the monologue whose outgoing audio is replaced by the prompt — the peer of that monologue hears it. Pass to_tag to scope to a specific peer in MPTY scenarios.

Requires rtpengine built with --with-transcoding and launched with --audio-player=on-demand. AMR-NB/WB prompts need licensed codec plugins; G.711 and Opus prompts work without them.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`file`	`Optional[str]`	Absolute path to an audio file on the rtpengine host.	`None`
`blob`	`Optional[bytes]`	Raw audio bytes to play (e.g. TTS output).	`None`
`db_id`	`Optional[int]`	Reference to a prompt stored in rtpengine's prompt DB.	`None`
`repeat`	`Optional[int]`	Number of times to repeat the prompt.	`None`
`start_ms`	`Optional[int]`	Offset into the file at which to start (ms).	`None`
`duration_ms`	`Optional[int]`	Cap on playback length (ms).	`None`
`to_tag`	`Optional[str]`	Optional peer tag for MPTY scoping.	`None`
`wait`	`bool`	When `True` (default, native siphon-rtp backend), the real runtime blocks until the prompt finishes playing so a script can sequence a following action (e.g. `echo()`) after it. The coroutine parks while it waits. `wait=False` returns as soon as the engine accepts the prompt (fire-and-forget). In this mock the call always returns immediately (the completion event is a runtime behavior); `wait` is recorded for assertions.	`True`

Returns:

Type	Description
`Optional[int]`	Prompt duration in ms if rtpengine reports one (mock returns
`Optional[int]`	the value set via :meth:`set_play_media_duration`, else `None`).

Example::

@b2bua.on_invite
async def on_invite(call):
    await rtpengine.offer(call, profile="ivr")
    call.answer(200, "OK", body=call.body, content_type="application/sdp")
    await rtpengine.play_media(call, file="/prompts/welcome.wav")  # wait=True
    await rtpengine.echo(call)                                     # after prompt

stop_media `async` ¶

stop_media(target: Any) -> bool

Stop any prompt currently playing on the selected monologue.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required

Returns:

Type	Description
`bool`	`True` on success.

play_dtmf `async` ¶

play_dtmf(
    target: Any,
    code: str,
    duration_ms: Optional[int] = None,
    volume_dbm0: Optional[int] = None,
    pause_ms: Optional[int] = None,
    to_tag: Optional[str] = None,
) -> bool

Inject DTMF tone(s) into the call.

Parameters:

Name	Type	Description	Default
`target`	`Any`	Request, Reply, or Call object.	required
`code`	`str`	A single digit (`"0"`–`"9"`, `"*"`, `"#"`, `"A"`–`"D"`) or a string sequence of digits.	required
`duration_ms`	`Optional[int]`	Tone duration per digit.	`None`
`volume_dbm0`	`Optional[int]`	Tone volume in dBm0 (typically `-8`).	`None`
`pause_ms`	`Optional[int]`	Inter-tone gap when `code` is a sequence.	`None`
`to_tag`	`Optional[str]`	Optional peer tag for MPTY scoping.	`None`

Example::

await rtpengine.play_dtmf(call, "123#", duration_ms=100)

silence_media `async` ¶

silence_media(target: Any) -> bool

Replace outgoing audio on the selected monologue with silence.

Pair with :meth:unsilence_media to restore the original stream.

unsilence_media `async` ¶

unsilence_media(target: Any) -> bool

Stop replacing outgoing audio with silence (undo :meth:silence_media).

block_media `async` ¶

block_media(target: Any) -> bool

Drop outgoing packets on the selected monologue entirely.

Pair with :meth:unblock_media to resume.

unblock_media `async` ¶

unblock_media(target: Any) -> bool

Resume forwarding the selected monologue's packets.

echo `async` ¶

echo(target: Any, enabled: bool = True) -> bool

Toggle echo-test mode on a call — reflect the caller's ingress audio back to itself (single-leg IVR echo).

enabled=False stops echoing. Native siphon-rtp backend only; DTMF and media-timeout events still fire while echoing.

subscribe_request `async` ¶

subscribe_request(
    call_id: str,
    from_tag: str,
    to_tag: str,
    sdp: Optional[bytes] = None,
    profile: Optional[str] = None,
) -> bytes

Create a new subscription to an existing call's media (MPTY / MRF conference focus).

Parameters:

Name	Type	Description	Default
`call_id`	`str`	rtpengine call-id of the source session.	required
`from_tag`	`str`	source monologue tag whose outgoing audio is subscribed.	required
`to_tag`	`str`	subscriber tag to create.	required
`sdp`	`Optional[bytes]`	Optional inbound SDP for the subscriber.	`None`
`profile`	`Optional[str]`	RTP profile name (defaults to `"rtp_passthrough"`).	`None`

Returns:

Type	Description
`bytes`	The subscriber SDP as `bytes`.

subscribe_answer `async` ¶

subscribe_answer(
    call_id: str,
    from_tag: str,
    to_tag: str,
    sdp: bytes,
    profile: Optional[str] = None,
) -> bytes

Complete the SDP negotiation for a subscription created via :meth:subscribe_request.

Returns:

Type	Description
`bytes`	The rewritten SDP as `bytes` (may be empty).

unsubscribe `async` ¶

unsubscribe(
    call_id: str, from_tag: str, to_tag: str
) -> bool

Tear down a subscription created via :meth:subscribe_request.

on_dtmf ¶

on_dtmf(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for inbound DTMF events from rtpengine.

Usage::

@rtpengine.on_dtmf
def handle_any(call_id, from_tag, digit, duration_ms, volume):
    ...

@rtpengine.on_dtmf(call_id="abc", from_tag="ftag1")
def handle_specific(call_id, from_tag, digit, duration_ms, volume):
    ...

fire_dtmf ¶

fire_dtmf(
    call_id: str,
    from_tag: str,
    digit: str,
    duration_ms: int = 0,
    volume: int = 0,
) -> int

Test helper: fire a DTMF event. Returns the number of handlers that matched (and were invoked).

on_media_timeout ¶

on_media_timeout(
    func_or_none: Any = None,
    *,
    call_id: Optional[str] = None,
    from_tag: Optional[str] = None
) -> Any

Register a handler for media-timeout events from the media engine.

The engine reaps a call whose media went dead and pushes a media-timeout event; the handler releases the per-call state no BYE will now clear (Rx/N5 QoS, charging, dialog).

Usage::

@rtpengine.on_media_timeout
def handle_any(call_id, from_tag):
    ...

@rtpengine.on_media_timeout(call_id="abc", from_tag="ftag1")
def handle_specific(call_id, from_tag):
    ...

fire_media_timeout ¶

fire_media_timeout(call_id: str, from_tag: str) -> int

Test helper: fire a media-timeout event. Returns the number of handlers that matched (and were invoked).

set_subscribe_request_sdp ¶

set_subscribe_request_sdp(sdp: bytes) -> None

Configure the SDP returned by :meth:subscribe_request (test helper).

set_subscribe_answer_sdp ¶

set_subscribe_answer_sdp(sdp: bytes) -> None

Configure the SDP returned by :meth:subscribe_answer (test helper).

set_play_media_duration ¶

set_play_media_duration(duration_ms: Optional[int]) -> None

Configure the duration returned by :meth:play_media (test helper).

set_answer_local_sdp ¶

set_answer_local_sdp(sdp: str) -> None

Configure the answer SDP returned by :meth:answer_local (test helper).

set_answer_local_no_codec ¶

set_answer_local_no_codec(no_codec: bool = True) -> None

Prime :meth:answer_local to model a no-encodable-codec offer (test helper) — the next answer_local records a deferred 488 (auto-reject) or raises ValueError (auto_reject=False).

clear ¶

clear() -> None

Clear recorded operations and registered event handlers (test helper).

`qos` namespace¶

Mock qos namespace — turns SDP offer/answer pairs into the media_components structure consumed by diameter.rx_aar and sbi.create_session.

The mock parses SDP just enough to produce a usable media_components list with RTP + RTCP sub-components for each m= section. Disabled streams (port 0) are skipped and a=rtcp-mux collapses RTCP into the RTP sub-component.

Example::

from siphon import qos, diameter

components = qos.media_flows_from_sdp(
    offer=request.body, answer=reply.body, direction="orig",
)
diameter.rx_aar(framed_ip=request.source_ip, media_components=components)

media_flows_from_sdp ¶

media_flows_from_sdp(
    *, offer: Any, answer: Any, direction: str = "orig"
) -> list[dict]

Translate an SDP offer/answer pair into a media_components list.

Parameters:

Name	Type	Description	Default
`offer`	`Any`	the original (offer) SDP — `str`, `bytes`, or a Request/Reply/Call mock with a `body` attribute.	required
`answer`	`Any`	the answer SDP (typically post `rtpengine.answer()`).	required
`direction`	`str`	`"orig"` (UE is offerer — UE addr from `offer`, remote from `answer`) or `"term"` (UE is answerer — addresses flipped).	`'orig'`

Returns:

Type	Description
`list[dict]`	list[dict]: one entry per non-disabled `m=` section.

Media¶

rtpengine namespace¶

operations instance-attribute ¶

media_calls instance-attribute ¶

active_sessions property ¶

instance_count property ¶

offer async ¶

answer async ¶

answer_local async ¶

delete async ¶

ping async ¶

play_media async ¶

stop_media async ¶

play_dtmf async ¶

silence_media async ¶

unsilence_media async ¶

block_media async ¶

unblock_media async ¶

echo async ¶

subscribe_request async ¶

subscribe_answer async ¶

unsubscribe async ¶

on_dtmf ¶

fire_dtmf ¶

on_media_timeout ¶

fire_media_timeout ¶

set_subscribe_request_sdp ¶

set_subscribe_answer_sdp ¶

set_play_media_duration ¶

set_answer_local_sdp ¶

set_answer_local_no_codec ¶

clear ¶

qos namespace¶

media_flows_from_sdp ¶

`rtpengine` namespace¶

operations `instance-attribute` ¶

media_calls `instance-attribute` ¶

active_sessions `property` ¶

instance_count `property` ¶

offer `async` ¶

answer `async` ¶

answer_local `async` ¶

delete `async` ¶

ping `async` ¶

play_media `async` ¶

stop_media `async` ¶

play_dtmf `async` ¶

silence_media `async` ¶

unsilence_media `async` ¶

block_media `async` ¶

unblock_media `async` ¶

echo `async` ¶

subscribe_request `async` ¶

subscribe_answer `async` ¶

unsubscribe `async` ¶

`qos` namespace¶