1use crate::llm::actions::{
2 client_trait::{Client, ClientActionResult},
3 protocol_trait::Protocol,
4 ActionDefinition, ParameterDefinition,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::srt::actions::{action, parameter};
8use anyhow::{bail, ensure, Result};
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct SrtClientProtocol;
14impl SrtClientProtocol {
15 pub fn new() -> Self {
16 Self
17 }
18}
19
20fn receive() -> ActionDefinition {
21 action(
22 "srt_receive",
23 "Read what the listener sends for a while and report it: messages, bytes, MPEG-TS packets with their PIDs and stream types, text messages, and SRT loss and retransmission statistics",
24 vec![parameter("seconds", "number", "How long to read, 1 to 60 (default 5)", false)],
25 json!({"type": "srt_receive", "seconds": 5}),
26 )
27}
28fn send_file() -> ActionDefinition {
29 action(
30 "srt_send_file",
31 "Send an MPEG-TS file the operator supplied, 7 TS packets (1316 bytes) per SRT message, paced at a bitrate",
32 vec![
33 parameter("path", "string", "Path of a .ts file under the client's media_root directory (64 MiB at most); a relative path is resolved against that directory, and nothing outside it is read", true),
34 parameter("bitrate_kbps", "number", "Pacing bitrate in kbit/s, 100 to 50000 (default 2000)", false),
35 ],
36 json!({"type": "srt_send_file", "path": "clip.ts", "bitrate_kbps": 2000}),
37 )
38}
39fn send_text() -> ActionDefinition {
40 action(
41 "srt_send_text",
42 "Send a UTF-8 text message as one SRT message",
43 vec![parameter(
44 "text",
45 "string",
46 "The message, up to 1316 bytes",
47 true,
48 )],
49 json!({"type": "srt_send_text", "text": "hello"}),
50 )
51}
52fn disconnect() -> ActionDefinition {
53 action(
54 "disconnect",
55 "Close the SRT connection",
56 vec![],
57 json!({"type": "disconnect"}),
58 )
59}
60fn actions() -> Vec<ActionDefinition> {
61 vec![receive(), send_file(), send_text(), disconnect()]
62}
63
64pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
65 EventType::new(
66 "srt_connected",
67 "The listener accepted the connection",
68 receive().example.clone(),
69 )
70 .with_parameters(vec![
71 parameter(
72 "stream_id",
73 "string",
74 "The stream ID this client sent",
75 true,
76 ),
77 parameter(
78 "latency_ms",
79 "number",
80 "The negotiated receive latency",
81 true,
82 ),
83 ])
84 .with_actions(actions())
85});
86pub static REPORT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
87 EventType::new(
88 "srt_report",
89 "What a receive or send did",
90 disconnect().example.clone(),
91 )
92 .with_parameters(vec![
93 parameter(
94 "operation",
95 "string",
96 "receive, send_file or send_text",
97 true,
98 ),
99 parameter("messages", "number", "SRT messages received or sent", true),
100 parameter("bytes", "number", "Payload bytes received or sent", true),
101 parameter(
102 "ts_packets",
103 "number",
104 "MPEG-TS packets (188 bytes, sync byte 0x47) among them",
105 false,
106 ),
107 parameter("pids", "array", "MPEG-TS PIDs seen", false),
108 parameter(
109 "stream_types",
110 "array",
111 "Elementary stream types from the PMT: h264, hevc, aac, mp3 or the hex code",
112 false,
113 ),
114 parameter("texts", "array", "Text messages received (up to 20)", false),
115 parameter(
116 "statistics",
117 "object",
118 "SRT statistics: packets, bytes, lost, retransmitted, dropped, RTT",
119 true,
120 ),
121 ])
122 .with_actions(actions())
123});
124
125impl Protocol for SrtClientProtocol {
126 fn protocol_name(&self) -> &'static str {
127 "SRT"
128 }
129 fn stack_name(&self) -> &'static str {
130 "ETH>IP>UDP>SRT"
131 }
132 fn keywords(&self) -> Vec<&'static str> {
133 vec!["srt", "secure reliable transport", "caller", "mpeg-ts"]
134 }
135 fn description(&self) -> &'static str {
136 "SRT caller: connects with a stream ID, reads and reports a stream, or sends an MPEG-TS file or text"
137 }
138 fn get_async_actions(&self, _: &crate::state::app_state::AppState) -> Vec<ActionDefinition> {
139 actions()
140 }
141 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
142 vec![]
143 }
144 fn get_event_types(&self) -> Vec<EventType> {
145 vec![CONNECTED_EVENT.clone(), REPORT_EVENT.clone()]
146 }
147 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
148 let p =
149 |name: &str, kind: &str, description: &str, example: Value, default: Option<Value>| {
150 ParameterDefinition {
151 name: name.into(),
152 type_hint: kind.into(),
153 description: description.into(),
154 required: false,
155 example,
156 default,
157 }
158 };
159 vec![
160 p(
161 "stream_id",
162 "string",
163 "Stream ID to send, e.g. #!::r=live/cam,m=request or publish:live/cam",
164 json!("#!::r=live/cam,m=publish"),
165 None,
166 ),
167 p(
168 "latency_ms",
169 "integer",
170 "SRT latency in milliseconds, 20 to 8000",
171 json!(200),
172 Some(json!(crate::server::srt::DEFAULT_LATENCY.as_millis() as u64)),
173 ),
174 p(
175 "passphrase",
176 "string",
177 "AES-128 passphrase (10 to 79 characters) when the listener encrypts",
178 json!("correct horse battery"),
179 None,
180 ),
181 p(
182 crate::client::media_root::MEDIA_ROOT_PARAM,
183 "string",
184 "Directory srt_send_file may read files from; nothing outside it is sent. Defaults to NetGet's own media directory under the platform's local-data dir.",
185 json!("/var/lib/netget/media"),
186 None,
187 ),
188 ]
189 }
190 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
191 use crate::protocol::metadata::*;
192 ProtocolMetadataV2::builder()
193 .state(DevelopmentState::Experimental)
194 .privilege_requirement(PrivilegeRequirement::None)
195 .implementation("srt-tokio 0.4.4 caller; MPEG-TS PAT/PMT inspection for reports")
196 .llm_control("What to read or send and when")
197 .e2e_testing("tests/client/srt: libsrt 1.5 (srt-live-transmit, independent) as listener sends a paced H.264/AAC MPEG-TS stream the client reads and reports, and writes the file the client publishes, which ffprobe decodes")
198 .notes("Live mode, caller only; no rendezvous. Files are paced by bitrate, not PCR.")
199 .max_inbound_bytes(1500)
200 .build()
201 }
202 fn example_prompt(&self) -> &'static str {
203 "Read live/cam from the SRT listener at 127.0.0.1:9000 for five seconds"
204 }
205 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
206 let llm = json!({"type":"open_client","protocol":"srt","remote_addr":"127.0.0.1:9000","instruction":"Read the stream for five seconds","startup_params":{"stream_id":"#!::r=live/cam,m=request"}});
207 let mut static_example = llm.clone();
208 static_example["event_handlers"] = json!([
209 {"event_pattern":"srt_connected","handler":{"type":"static","actions":[receive().example]}},
210 {"event_pattern":"srt_report","handler":{"type":"static","actions":[{"type":"disconnect"}]}}
211 ]);
212 let mut scripted = static_example.clone();
213 scripted["event_handlers"][1]["handler"] = json!({"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'disconnect'}] if e['bytes']>0 else [{'type':'srt_receive','seconds':5}]}))"});
214 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
215 }
216 fn group_name(&self) -> &'static str {
217 "Real-time"
218 }
219}
220
221impl Client for SrtClientProtocol {
222 fn connect(
223 &self,
224 ctx: ConnectContext,
225 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
226 {
227 Box::pin(super::connect(ctx))
228 }
229 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
230 match v["type"].as_str() {
231 Some("srt_receive") => {
232 if let Some(s) = v.get("seconds").filter(|s| !s.is_null()) {
233 ensure!(
234 s.as_f64().is_some_and(|s| (1.0..=60.0).contains(&s)),
235 "seconds is 1 to 60"
236 );
237 }
238 }
239 Some("srt_send_file") => {
240 ensure!(
241 v["path"].as_str().is_some_and(|p| !p.is_empty()
242 && p.len() <= 1024
243 && !crate::utils::sanitize::has_controls(&p)),
244 "path is a file path"
245 );
246 if let Some(b) = v.get("bitrate_kbps").filter(|b| !b.is_null()) {
247 ensure!(
248 b.as_f64().is_some_and(|b| (100.0..=50_000.0).contains(&b)),
249 "bitrate_kbps is 100 to 50000"
250 );
251 }
252 }
253 Some("srt_send_text") => ensure!(
254 v["text"]
255 .as_str()
256 .is_some_and(|t| !t.is_empty() && t.len() <= 1316),
257 "text is 1 to 1316 bytes"
258 ),
259 Some("disconnect") => return Ok(ClientActionResult::Disconnect),
260 _ => bail!("Unknown SRT client action"),
261 }
262 Ok(ClientActionResult::Custom {
263 name: v["type"].as_str().unwrap_or_default().into(),
264 data: v,
265 })
266 }
267}