1use crate::llm::actions::{
5 client_trait::{Client, ClientActionResult},
6 protocol_trait::Protocol,
7 ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
8};
9use crate::protocol::log_template::LogTemplate;
10use crate::protocol::{ConnectContext, EventType};
11use crate::state::app_state::AppState;
12use anyhow::{bail, ensure, Result};
13use serde_json::{json, Value};
14use std::sync::LazyLock;
15
16pub const OPTIONS: &str = "rtsp_options";
17pub const DESCRIBE: &str = "rtsp_describe";
18pub const SETUP: &str = "rtsp_setup";
19pub const PLAY: &str = "rtsp_play";
20pub const PAUSE: &str = "rtsp_pause";
21pub const TEARDOWN: &str = "rtsp_teardown";
22
23#[derive(Default)]
24pub struct RtspClientProtocol;
25impl RtspClientProtocol {
26 pub fn new() -> Self {
27 Self
28 }
29}
30
31fn p(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
32 Parameter {
33 name: name.into(),
34 type_hint: kind.into(),
35 description: description.into(),
36 required,
37 }
38}
39
40fn action(
41 name: &str,
42 description: &str,
43 parameters: Vec<Parameter>,
44 example: Value,
45) -> ActionDefinition {
46 ActionDefinition {
47 name: name.into(),
48 description: description.into(),
49 parameters,
50 example,
51 log_template: Some(LogTemplate::new().with_info(format!("-> RTSP {name}"))),
52 }
53}
54
55pub fn actions() -> Vec<ActionDefinition> {
56 vec![
57 action(OPTIONS, "Ask the server which methods it supports (OPTIONS).", vec![], json!({"type": OPTIONS})),
58 action(DESCRIBE, "Ask for the stream's description (DESCRIBE); the SDP comes back parsed.", vec![], json!({"type": DESCRIBE})),
59 action(SETUP, "Set a track up over UDP (SETUP). Rust binds the RTP/RTCP ports and keeps the session.",
60 vec![p("track", "string", "The track's control URL from the DESCRIBE answer (absolute or relative); default the first audio or video track", false)],
61 json!({"type": SETUP})),
62 action(PLAY, "Start the media of the session (PLAY); what arrives is summarised per stream.",
63 vec![p("range", "string", "An RTSP Range, e.g. npt=0- (default none)", false)], json!({"type": PLAY})),
64 action(PAUSE, "Pause the session's media (PAUSE).", vec![], json!({"type": PAUSE})),
65 action(TEARDOWN, "End the session (TEARDOWN); its RTP ports close.", vec![], json!({"type": TEARDOWN})),
66 action("disconnect", "Close the RTSP connection.", vec![], json!({"type": "disconnect"})),
67 ]
68}
69
70fn event(id: &str, description: &str, params: Vec<Parameter>) -> EventType {
71 EventType::new(id, description, json!({"type": DESCRIBE}))
72 .with_parameters(params)
73 .with_actions(actions())
74}
75
76pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
77 event(
78 "rtsp_connected",
79 "Connected to the RTSP server.",
80 vec![p(
81 "url",
82 "string",
83 "The stream URL requests are made for",
84 true,
85 )],
86 )
87});
88
89pub static RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
90 event("rtsp_response", "The server answered a request.", vec![
91 p("method", "string", "The request answered, e.g. DESCRIBE", true),
92 p("status", "number", "RTSP status code, e.g. 200 or 404", true),
93 p("reason", "string", "The status line's reason phrase", true),
94 p("headers", "object", "Public, Content-Base, Session, Transport, RTP-Info and Range when present", true),
95 p("sdp", "object", "For DESCRIBE: {media: [{type, port, protocol, formats, rtpmap, control}], control}", false),
96 p("error", "string", "Why the request failed before an answer", false),
97 ])
98});
99
100pub static STREAM_STARTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
101 event(
102 "rtsp_stream_started",
103 "RTP began arriving on a set-up track.",
104 vec![
105 p(
106 "ssrc",
107 "number",
108 "The stream's synchronisation source",
109 true,
110 ),
111 p("payload_type", "number", "Its RTP payload type", true),
112 p(
113 "codec",
114 "string",
115 "pcmu or pcma when it is G.711, otherwise unknown",
116 true,
117 ),
118 p("from", "string", "The server's RTP address", true),
119 ],
120 )
121});
122
123pub static STREAM_ENDED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
124 event("rtsp_stream_ended", "A stream stopped (nothing for a second): what arrived, and for G.711 what it sounded like.", vec![
125 p("ssrc", "number", "The stream's synchronisation source", true),
126 p("packets", "number", "RTP packets received", true),
127 p("lost", "number", "Packets missing from the sequence numbers", true),
128 p("octets", "number", "Payload bytes received", true),
129 p("duration_ms", "number", "Media length by RTP timestamps at 8 kHz", true),
130 p("tone_hz", "number", "For G.711: the dominant frequency of the decoded audio", false),
131 p("level_dbfs", "number", "For G.711: the decoded audio's RMS level in dBFS", false),
132 ])
133});
134
135pub fn check(v: &Value) -> Result<()> {
136 match v["type"].as_str().unwrap_or_default() {
137 OPTIONS | DESCRIBE | PAUSE | TEARDOWN => {}
138 SETUP => {
139 if let Some(t) = v.get("track").filter(|x| !x.is_null()) {
140 let t = t.as_str().unwrap_or_default();
141 ensure!(
142 !t.is_empty() && t.len() <= 1024 && !t.contains(['\r', '\n', ' ']),
143 "track is a control URL"
144 );
145 }
146 }
147 PLAY => {
148 if let Some(r) = v.get("range").filter(|x| !x.is_null()) {
149 let r = r.as_str().unwrap_or_default();
150 ensure!(
151 r.len() <= 128 && !r.contains(['\r', '\n']),
152 "range is an RTSP Range value"
153 );
154 }
155 }
156 other => bail!("Unknown RTSP client action {other:?}"),
157 }
158 Ok(())
159}
160
161impl Protocol for RtspClientProtocol {
162 fn protocol_name(&self) -> &'static str {
163 "RTSP"
164 }
165 fn stack_name(&self) -> &'static str {
166 "ETH>IP>TCP>RTSP"
167 }
168 fn keywords(&self) -> Vec<&'static str> {
169 vec![
170 "rtsp client",
171 "rtsp",
172 "rtsp player",
173 "ip camera",
174 "streaming client",
175 ]
176 }
177 fn description(&self) -> &'static str {
178 "RTSP client: describes, sets up and plays streams from a media server and reports the media that arrives"
179 }
180 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
181 actions()
182 }
183 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
184 vec![]
185 }
186 fn get_event_types(&self) -> Vec<EventType> {
187 vec![
188 CONNECTED_EVENT.clone(),
189 RESPONSE_EVENT.clone(),
190 STREAM_STARTED_EVENT.clone(),
191 STREAM_ENDED_EVENT.clone(),
192 ]
193 }
194 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
195 vec![]
196 }
197 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
198 use crate::protocol::metadata::*;
199 ProtocolMetadataV2::builder()
200 .state(DevelopmentState::Experimental)
201 .privilege_requirement(PrivilegeRequirement::None)
202 .implementation("RTSP/1.0 over Tokio TCP (CSeq, Session, Content-Base and Transport handled in Rust), SDP parsed for the model, UDP RTP/RTCP port pairs bound per SETUP, and the RTP client's per-stream tracker (loss, length, decoded G.711 tone and level)")
203 .llm_control("Which requests to make in which order, which track to set up, and what to do with the media that arrives")
204 .e2e_testing("tests/client/rtsp: mediamtx 1.9.3 serving a G.711 stream that ffmpeg publishes into it; NetGet describes, sets up, plays and tears down, and the tone it decodes is asserted")
205 .notes("UDP transport only (no TCP interleaving), no authentication, no RTSP 2.0, no requests from the server. A handler chain stops after 8 follow-ups.")
206 .build()
207 }
208 fn example_prompt(&self) -> &'static str {
209 "Open rtsp://127.0.0.1:8554/cam, describe it, play its audio for a few seconds and tell me what it sounds like"
210 }
211 fn get_startup_examples(&self) -> StartupExamples {
212 let llm = json!({"type":"open_client","protocol":"rtsp","remote_addr":"rtsp://127.0.0.1:8554/cam",
213 "instruction":"Describe the stream, set up its first track and play it"});
214 let mut static_example = llm.clone();
215 static_example["event_handlers"] = json!([{"event_pattern":"rtsp_connected","handler":{"type":"static","actions":[{"type":DESCRIBE}]}}]);
216 let mut scripted = llm.clone();
217 scripted["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python",
218 "code":"import json,sys\ni=json.load(sys.stdin); t=i['event_type_id']; e=i['event']\nnext={'DESCRIBE':'rtsp_setup','SETUP':'rtsp_play'}\na=[{'type':'rtsp_describe'}] if t=='rtsp_connected' else ([{'type':next[e['method']]}] if t=='rtsp_response' and e['status']==200 and e['method'] in next else [])\nprint(json.dumps({'actions':a}))"}}]);
219 StartupExamples::new(llm, scripted, static_example)
220 }
221 fn group_name(&self) -> &'static str {
222 "Proxy & Network"
223 }
224}
225
226impl Client for RtspClientProtocol {
227 fn connect(
228 &self,
229 ctx: ConnectContext,
230 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
231 {
232 Box::pin(super::connect(ctx))
233 }
234 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
235 let name = v["type"].as_str().unwrap_or_default().to_string();
236 if name == "disconnect" {
237 return Ok(ClientActionResult::Disconnect);
238 }
239 check(&v)?;
240 Ok(ClientActionResult::Custom { name, data: v })
241 }
242}