netget/client/rtp/
actions.rs

1//! What the model does as an RTP endpoint: stream synthesized G.711 audio to the remote
2//! (described, never sampled), report as a sender over RTCP, and hear about the streams that
3//! arrive — one event when a stream starts and one when it ends, never one per packet.
4use crate::llm::actions::{
5    client_trait::{Client, ClientActionResult},
6    protocol_trait::Protocol,
7    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
8};
9use crate::protocol::{ConnectContext, EventType};
10use crate::server::rtp::actions::{send_rtcp_sender_report_action, send_rtp_audio_action};
11use crate::server::rtp::media;
12use crate::state::app_state::AppState;
13use anyhow::{bail, Result};
14use serde_json::{json, Value};
15use std::sync::LazyLock;
16
17pub const SEND_AUDIO: &str = "send_rtp_audio";
18pub const SEND_SR: &str = "send_rtcp_sender_report";
19/// Where the endpoint receives when nowhere is named.
20pub const DEFAULT_LISTEN: &str = "127.0.0.1:0";
21
22#[derive(Default)]
23pub struct RtpClientProtocol;
24impl RtpClientProtocol {
25    pub fn new() -> Self {
26        Self
27    }
28}
29
30fn p(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
31    Parameter {
32        name: name.into(),
33        type_hint: kind.into(),
34        description: description.into(),
35        required,
36    }
37}
38
39pub fn actions() -> Vec<ActionDefinition> {
40    let mut audio = send_rtp_audio_action();
41    audio.description = "Stream synthesized G.711 audio to the remote as paced 20 ms RTP packets (one stream at a time; a new one replaces it). Describe the content; Rust owns the samples and the framing.".into();
42    let mut sr = send_rtcp_sender_report_action();
43    sr.description = "Send the remote a minimal RTCP Sender Report (RFC 3550 §6.4.1).".into();
44    vec![
45        audio,
46        sr,
47        ActionDefinition {
48            name: "disconnect".into(),
49            description: "Stop: any stream being sent stops and the socket closes.".into(),
50            parameters: vec![],
51            example: json!({"type": "disconnect"}),
52            log_template: Some(
53                crate::protocol::log_template::LogTemplate::new().with_info("-> RTP disconnect"),
54            ),
55        },
56    ]
57}
58
59fn event(id: &str, description: &str, params: Vec<Parameter>) -> EventType {
60    EventType::new(id, description, json!({"type": SEND_AUDIO, "payload_type": "pcmu", "content": "tone", "tone_hz": 440, "duration_ms": 1000}))
61        .with_parameters(params)
62        .with_actions(actions())
63}
64
65pub static READY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
66    event(
67        "rtp_ready",
68        "The endpoint is bound; nothing is sent until an action asks.",
69        vec![
70            p(
71                "local_addr",
72                "string",
73                "Where this endpoint receives RTP",
74                true,
75            ),
76            p("remote_addr", "string", "Where it sends RTP", true),
77        ],
78    )
79});
80
81pub static STREAM_STARTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
82    event(
83        "rtp_stream_started",
84        "A new RTP stream (a new SSRC) began arriving.",
85        vec![
86            p(
87                "ssrc",
88                "number",
89                "The stream's synchronisation source",
90                true,
91            ),
92            p("payload_type", "number", "Its RTP payload type", true),
93            p(
94                "codec",
95                "string",
96                "pcmu or pcma when it is G.711, otherwise unknown",
97                true,
98            ),
99            p("from", "string", "The sender's address and port", true),
100        ],
101    )
102});
103
104pub static STREAM_ENDED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
105    event("rtp_stream_ended", "A stream stopped (nothing for a second): what arrived, and for G.711 what it sounded like.", vec![
106        p("ssrc", "number", "The stream's synchronisation source", true),
107        p("packets", "number", "Packets received", true),
108        p("lost", "number", "Packets missing from the sequence numbers", true),
109        p("octets", "number", "Payload bytes received", true),
110        p("duration_ms", "number", "Media length by RTP timestamps at 8 kHz", true),
111        p("tone_hz", "number", "For G.711: the dominant frequency of the decoded audio, estimated from zero crossings", false),
112        p("level_dbfs", "number", "For G.711: the decoded audio's RMS level in dBFS", false),
113    ])
114});
115
116pub static SENT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
117    event(
118        "rtp_sent",
119        "A stream this endpoint was sending finished (or was replaced).",
120        vec![
121            p("ssrc", "number", "The SSRC it was sent with", true),
122            p(
123                "packets",
124                "number",
125                "How many RTP packets went out, one per 20 ms frame",
126                true,
127            ),
128            p(
129                "duration_ms",
130                "number",
131                "How much audio went out, in milliseconds",
132                true,
133            ),
134            p(
135                "complete",
136                "boolean",
137                "False when a newer stream or a disconnect cut it short",
138                true,
139            ),
140        ],
141    )
142});
143
144pub static RTCP_EVENT: LazyLock<EventType> = LazyLock::new(|| {
145    event(
146        "rtcp_received",
147        "An RTCP packet arrived (sender or receiver report, BYE, …).",
148        vec![
149            p(
150                "packet_type",
151                "number",
152                "200 SR, 201 RR, 202 SDES, 203 BYE, 204 APP",
153                true,
154            ),
155            p("ssrc", "number", "The reporting source", true),
156            p("from", "string", "The sender's address and port", true),
157        ],
158    )
159});
160
161pub fn check(v: &Value) -> Result<()> {
162    match v["type"].as_str().unwrap_or_default() {
163        SEND_AUDIO => {
164            let codec = v
165                .get("payload_type")
166                .and_then(Value::as_str)
167                .map(media::AudioCodec::parse)
168                .unwrap_or(Ok(media::AudioCodec::Pcmu))?;
169            let content = media::parse_audio_content(v)?;
170            media::synthesize(codec, &content, v["duration_ms"].as_u64().unwrap_or(1000))?;
171            Ok(())
172        }
173        SEND_SR => Ok(()),
174        other => bail!("Unknown RTP client action {other:?}"),
175    }
176}
177
178impl Protocol for RtpClientProtocol {
179    fn protocol_name(&self) -> &'static str {
180        "RTP"
181    }
182    fn stack_name(&self) -> &'static str {
183        "ETH>IP>UDP>RTP"
184    }
185    fn keywords(&self) -> Vec<&'static str> {
186        vec![
187            "rtp client",
188            "rtp sender",
189            "rtp",
190            "voip audio",
191            "real-time transport",
192        ]
193    }
194    fn description(&self) -> &'static str {
195        "RTP endpoint that streams synthesized G.711 to a remote and reports the streams it receives"
196    }
197    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
198        actions()
199    }
200    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
201        vec![]
202    }
203    fn get_event_types(&self) -> Vec<EventType> {
204        vec![
205            READY_EVENT.clone(),
206            STREAM_STARTED_EVENT.clone(),
207            STREAM_ENDED_EVENT.clone(),
208            SENT_EVENT.clone(),
209            RTCP_EVENT.clone(),
210        ]
211    }
212    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
213        vec![ParameterDefinition {
214            name: "listen".into(),
215            type_hint: "string".into(),
216            description: "Address this endpoint receives RTP on (the remote sends media here)"
217                .into(),
218            required: false,
219            example: json!("0.0.0.0:40002"),
220            default: Some(json!(DEFAULT_LISTEN)),
221        }]
222    }
223    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
224        use crate::protocol::metadata::*;
225        ProtocolMetadataV2::builder()
226            .state(DevelopmentState::Experimental)
227            .privilege_requirement(PrivilegeRequirement::None)
228            .implementation("Tokio UDP with the RTP server's synthesis and packetizer (G.711 PCMU/PCMA, 20 ms frames paced in real time); inbound RTP tracked per SSRC (loss from sequence numbers, length from timestamps) and G.711 decoded to estimate tone and level; RTCP Sender Reports")
229            .llm_control("What audio to send and when, and what to do about the streams that arrive")
230            .e2e_testing("tests/client/rtp: ffmpeg receives NetGet's stream through an SDP and decodes it (its tone measured in the WAV it writes), and ffmpeg's own RTP muxer streams a sine to NetGet, whose decoded tone and packet accounting are asserted")
231            .notes("Audio only, G.711 only; no SRTP, no jitter buffer, no receiver reports. Inbound streams are summarised, never reported per packet.")
232            .build()
233    }
234    fn example_prompt(&self) -> &'static str {
235        "Send a 440 Hz tone for two seconds to the RTP endpoint at 127.0.0.1:40000"
236    }
237    fn get_startup_examples(&self) -> StartupExamples {
238        let llm = json!({"type":"open_client","protocol":"rtp","remote_addr":"127.0.0.1:40000",
239            "instruction":"Send a 440 Hz PCMU tone for two seconds, then report what comes back"});
240        let mut static_example = llm.clone();
241        static_example["event_handlers"] = json!([{"event_pattern":"rtp_ready","handler":{"type":"static","actions":[{"type":SEND_AUDIO,"content":"tone","tone_hz":440,"duration_ms":2000}]}}]);
242        let mut scripted = llm.clone();
243        scripted["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python",
244            "code":"import json,sys\ni=json.load(sys.stdin); t=i['event_type_id']\na=[{'type':'send_rtp_audio','content':'dtmf','digits':'123#'}] if t=='rtp_ready' else []\nprint(json.dumps({'actions':a}))"}}]);
245        StartupExamples::new(llm, scripted, static_example)
246    }
247    fn group_name(&self) -> &'static str {
248        "Proxy & Network"
249    }
250}
251
252impl Client for RtpClientProtocol {
253    fn connect(
254        &self,
255        ctx: ConnectContext,
256    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
257    {
258        Box::pin(super::connect(ctx))
259    }
260    fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
261        let name = v["type"].as_str().unwrap_or_default().to_string();
262        if name == "disconnect" {
263            return Ok(ClientActionResult::Disconnect);
264        }
265        check(&v)?;
266        Ok(ClientActionResult::Custom { name, data: v })
267    }
268}
netget/client/rtp/actions.rs

netget/client/rtp/
actions.rs
