openipc_web/
receiver.rs

1use js_sys::{Array, Object, Reflect, Uint8Array};
2use openipc_core::realtek::{parse_rx_aggregate_with_kind, RxDescriptorKind};
3use openipc_core::{
4    ChannelId, FrameLayout, PayloadRouteId, RadioPort, ReceiverBatch, ReceiverBatchOptions,
5    ReceiverRuntime, RtpPayloadTap, WfbKeypair,
6};
7use wasm_bindgen::prelude::*;
8
9use crate::js::{
10    counters_json, counters_object, elapsed_ms, now_ms, parse_hex_u64, raw_payload_object,
11    set_number,
12};
13use crate::video::{rtp_status_object, video_frame_object};
14
15const VIDEO_ROUTE_ID: PayloadRouteId = PayloadRouteId::new(1);
16const TELEMETRY_ROUTE_ID: PayloadRouteId = PayloadRouteId::new(2);
17const DEFAULT_KEY_SLOT: u64 = 0;
18
19#[wasm_bindgen]
20/// Browser/WASM receiver for OpenIPC RX transfers and RTP packets.
21pub struct OpenIpcReceiver {
22    pub(crate) runtime: ReceiverRuntime,
23    pub(crate) rx_descriptor_kind: RxDescriptorKind,
24}
25
26impl OpenIpcReceiver {
27    pub(crate) fn video_fec_counters(&self) -> openipc_core::FecCounters {
28        self.runtime.video_fec_counters()
29    }
30}
31
32#[wasm_bindgen]
33impl OpenIpcReceiver {
34    #[wasm_bindgen(constructor)]
35    /// Create a plain/FEC-only receiver for the default OpenIPC video channel.
36    pub fn new() -> Result<OpenIpcReceiver, JsValue> {
37        Self::with_channel_id(openipc_core::channel::DEFAULT_LINK_ID << 8, 1, 5)
38    }
39
40    #[wasm_bindgen(js_name = withChannelId)]
41    /// Create a plain/FEC-only receiver for a specific channel id.
42    pub fn with_channel_id(
43        channel_id: u32,
44        fec_k: usize,
45        fec_n: usize,
46    ) -> Result<OpenIpcReceiver, JsValue> {
47        let runtime = ReceiverRuntime::with_plain_video_route(
48            FrameLayout::WithFcs,
49            VIDEO_ROUTE_ID,
50            ChannelId::new(channel_id),
51            DEFAULT_KEY_SLOT,
52            fec_k,
53            fec_n,
54        )
55        .map_err(|err| JsValue::from_str(&format!("invalid receiver config: {err}")))?;
56        Ok(Self {
57            runtime,
58            rx_descriptor_kind: RxDescriptorKind::Jaguar1,
59        })
60    }
61
62    #[wasm_bindgen(js_name = withKeypair)]
63    /// Create an encrypted WFB receiver and default telemetry downlink tap.
64    pub fn with_keypair(
65        channel_id: u32,
66        keypair: &[u8],
67        minimum_epoch: u64,
68    ) -> Result<OpenIpcReceiver, JsValue> {
69        let keypair = WfbKeypair::from_bytes(keypair)
70            .map_err(|err| JsValue::from_str(&format!("invalid WFB keypair: {err}")))?;
71        let telemetry_channel_id =
72            ChannelId::from_link_port(channel_id >> 8, RadioPort::TelemetryRx).raw();
73        openipc_receiver_with_keypair_and_telemetry_channel_inner(
74            channel_id,
75            telemetry_channel_id,
76            keypair,
77            minimum_epoch,
78        )
79    }
80
81    #[wasm_bindgen(js_name = withKeypairOnly)]
82    /// Create an encrypted WFB receiver with only the video route.
83    pub fn with_keypair_only(
84        channel_id: u32,
85        keypair: &[u8],
86        minimum_epoch: u64,
87    ) -> Result<OpenIpcReceiver, JsValue> {
88        let keypair = WfbKeypair::from_bytes(keypair)
89            .map_err(|err| JsValue::from_str(&format!("invalid WFB keypair: {err}")))?;
90        let runtime = ReceiverRuntime::with_keyed_video_route(
91            FrameLayout::WithFcs,
92            VIDEO_ROUTE_ID,
93            ChannelId::new(channel_id),
94            DEFAULT_KEY_SLOT,
95            keypair,
96            minimum_epoch,
97        )
98        .map_err(|err| JsValue::from_str(&format!("invalid encrypted receiver config: {err}")))?;
99        Ok(OpenIpcReceiver {
100            runtime,
101            rx_descriptor_kind: RxDescriptorKind::Jaguar1,
102        })
103    }
104
105    #[wasm_bindgen(js_name = setRxDescriptorKind)]
106    /// Select the Realtek USB RX descriptor layout for future bulk-IN transfers.
107    pub fn set_rx_descriptor_kind(&mut self, kind: &str) -> Result<(), JsValue> {
108        self.rx_descriptor_kind = parse_rx_descriptor_kind(kind)?;
109        Ok(())
110    }
111
112    #[wasm_bindgen(js_name = setRtpReorderEnabled)]
113    /// Enable or disable the small RTP sequence reorder buffer.
114    pub fn set_rtp_reorder_enabled(&mut self, enabled: bool) {
115        self.runtime.set_rtp_reorder_enabled(enabled);
116    }
117
118    #[wasm_bindgen(js_name = withKeypairAndMavlinkChannel)]
119    /// Create an encrypted WFB receiver with an explicit raw telemetry channel.
120    ///
121    /// This is the historical JS name. New applications should call
122    /// `withKeypairAndTelemetryChannel`.
123    pub fn with_keypair_and_mavlink_channel(
124        channel_id: u32,
125        mavlink_channel_id: u32,
126        keypair: &[u8],
127        minimum_epoch: u64,
128    ) -> Result<OpenIpcReceiver, JsValue> {
129        Self::with_keypair_and_telemetry_channel(
130            channel_id,
131            mavlink_channel_id,
132            keypair,
133            minimum_epoch,
134        )
135    }
136
137    #[wasm_bindgen(js_name = withKeypairAndTelemetryChannel)]
138    /// Create an encrypted WFB receiver with an explicit raw telemetry channel.
139    pub fn with_keypair_and_telemetry_channel(
140        channel_id: u32,
141        telemetry_channel_id: u32,
142        keypair: &[u8],
143        minimum_epoch: u64,
144    ) -> Result<OpenIpcReceiver, JsValue> {
145        let keypair = WfbKeypair::from_bytes(keypair)
146            .map_err(|err| JsValue::from_str(&format!("invalid WFB keypair: {err}")))?;
147        openipc_receiver_with_keypair_and_telemetry_channel_inner(
148            channel_id,
149            telemetry_channel_id,
150            keypair,
151            minimum_epoch,
152        )
153    }
154
155    #[wasm_bindgen(js_name = pushRtpPacket)]
156    /// Push one raw RTP packet and return Annex-B bytes when a frame completes.
157    pub fn push_rtp_packet(&mut self, data: &[u8]) -> Option<Uint8Array> {
158        self.runtime
159            .push_rtp_packet(data)
160            .ok()
161            .and_then(|mut frames| frames.drain(..).next())
162            .map(|frame| Uint8Array::from(frame.data.as_slice()))
163    }
164
165    #[wasm_bindgen(
166        js_name = pushRtpPacketDetailed,
167        unchecked_return_type = "OpenIpcVideoFrame | null"
168    )]
169    /// Push one RTP packet and return a typed frame object when one completes.
170    pub fn push_rtp_packet_detailed(&mut self, data: &[u8]) -> Result<JsValue, JsValue> {
171        match self
172            .runtime
173            .push_rtp_packet(data)
174            .ok()
175            .and_then(|mut frames| frames.drain(..).next())
176        {
177            Some(frame) => Ok(video_frame_object(frame)?.into()),
178            None => Ok(JsValue::NULL),
179        }
180    }
181
182    #[wasm_bindgen(js_name = pushDecryptedFragment)]
183    /// Push an already-decrypted WFB fragment into the video runtime.
184    pub fn push_decrypted_fragment(
185        &mut self,
186        data_nonce_hex: &str,
187        fragment: &[u8],
188    ) -> Result<Array, JsValue> {
189        let data_nonce = parse_hex_u64(data_nonce_hex)?;
190        let batch = self
191            .runtime
192            .push_decrypted_fragment(
193                self.runtime.video_runtime(),
194                data_nonce,
195                fragment,
196                &ReceiverBatchOptions::default(),
197            )
198            .map_err(|err| JsValue::from_str(&format!("WFB fragment rejected: {err}")))?;
199        Ok(frame_bytes_array(batch.frames))
200    }
201
202    #[wasm_bindgen(js_name = pushDecrypted80211Frame)]
203    /// Push an 802.11 frame with a caller-supplied decrypted WFB fragment.
204    pub fn push_decrypted_80211_frame(
205        &mut self,
206        frame: &[u8],
207        fragment: &[u8],
208    ) -> Result<Array, JsValue> {
209        let batch = self
210            .runtime
211            .push_decrypted_80211_frame(
212                self.runtime.video_runtime(),
213                frame,
214                fragment,
215                &ReceiverBatchOptions::default(),
216            )
217            .map_err(|err| JsValue::from_str(&format!("802.11 frame rejected: {err}")))?;
218        Ok(frame_bytes_array(batch.frames))
219    }
220
221    #[wasm_bindgen(js_name = pushEncrypted80211Frame)]
222    /// Push one encrypted OpenIPC/WFB 802.11 frame.
223    pub fn push_encrypted_80211_frame(&mut self, frame: &[u8]) -> Result<Array, JsValue> {
224        let batch = self
225            .runtime
226            .push_80211_frame(frame, &ReceiverBatchOptions::default())
227            .map_err(|err| JsValue::from_str(&format!("802.11 frame rejected: {err}")))?;
228        Ok(frame_bytes_array(batch.frames))
229    }
230
231    #[wasm_bindgen(js_name = pushRxTransfer)]
232    /// Push one Realtek RX USB transfer and return completed Annex-B frames.
233    pub fn push_rx_transfer(&mut self, transfer: &[u8]) -> Result<Array, JsValue> {
234        let batch = self
235            .runtime
236            .push_rx_transfer_with_kind(
237                transfer,
238                self.rx_descriptor_kind,
239                &ReceiverBatchOptions::default(),
240            )
241            .map_err(|err| JsValue::from_str(&format!("Realtek RX aggregate rejected: {err}")))?;
242        Ok(frame_bytes_array(batch.frames))
243    }
244
245    #[wasm_bindgen(
246        js_name = pushRxTransferDetailed,
247        unchecked_return_type = "OpenIpcVideoFrame[]"
248    )]
249    /// Push one RX transfer and return typed frame objects.
250    pub fn push_rx_transfer_detailed(&mut self, transfer: &[u8]) -> Result<Array, JsValue> {
251        self.push_rx_transfer_detailed_with_options(transfer, false)
252    }
253
254    #[wasm_bindgen(
255        js_name = pushRxTransferDetailedWithOptions,
256        unchecked_return_type = "OpenIpcVideoFrame[]"
257    )]
258    /// Push one RX transfer with control over CRC/ICV-marked packets.
259    pub fn push_rx_transfer_detailed_with_options(
260        &mut self,
261        transfer: &[u8],
262        keep_corrupted: bool,
263    ) -> Result<Array, JsValue> {
264        let batch = self
265            .runtime
266            .push_rx_transfer_with_kind(
267                transfer,
268                self.rx_descriptor_kind,
269                &ReceiverBatchOptions {
270                    accept_corrupted: keep_corrupted,
271                    ..ReceiverBatchOptions::default()
272                },
273            )
274            .map_err(|err| JsValue::from_str(&format!("Realtek RX aggregate rejected: {err}")))?;
275        frame_objects_array(batch.frames)
276    }
277
278    #[wasm_bindgen(
279        js_name = pushRxTransferProfiled,
280        unchecked_return_type = "OpenIpcRxTransferProfile"
281    )]
282    /// Push one RX transfer and return frames plus parser/latency counters.
283    pub fn push_rx_transfer_profiled(&mut self, transfer: &[u8]) -> Result<Object, JsValue> {
284        self.push_rx_transfer_profiled_with_options(transfer, false)
285    }
286
287    #[wasm_bindgen(
288        js_name = pushRxTransferProfiledWithOptions,
289        unchecked_return_type = "OpenIpcRxTransferProfile"
290    )]
291    /// Push one RX transfer with profiling and bad-FCS handling options.
292    pub fn push_rx_transfer_profiled_with_options(
293        &mut self,
294        transfer: &[u8],
295        keep_corrupted: bool,
296    ) -> Result<Object, JsValue> {
297        self.push_rx_transfer_profiled_inner(
298            transfer,
299            keep_corrupted,
300            &[TELEMETRY_ROUTE_ID.raw() as u32],
301            &[],
302        )
303    }
304
305    #[wasm_bindgen(
306        js_name = pushRxTransferProfiledWithRouteIds,
307        unchecked_return_type = "OpenIpcRxTransferProfile"
308    )]
309    /// Push one RX transfer and copy raw payloads for caller-selected route IDs.
310    pub fn push_rx_transfer_profiled_with_route_ids(
311        &mut self,
312        transfer: &[u8],
313        keep_corrupted: bool,
314        raw_route_ids: &[u32],
315    ) -> Result<Object, JsValue> {
316        self.push_rx_transfer_profiled_inner(transfer, keep_corrupted, raw_route_ids, &[])
317    }
318
319    #[wasm_bindgen(
320        js_name = pushRxTransferProfiledWithRouteIdsAndRtpTaps,
321        unchecked_return_type = "OpenIpcRxTransferProfile"
322    )]
323    /// Push one RX transfer and copy raw payloads plus filtered RTP payload taps.
324    pub fn push_rx_transfer_profiled_with_route_ids_and_rtp_taps(
325        &mut self,
326        transfer: &[u8],
327        keep_corrupted: bool,
328        raw_route_ids: &[u32],
329        rtp_tap_route_ids: &[u32],
330        rtp_tap_payload_types: &[u8],
331    ) -> Result<Object, JsValue> {
332        if rtp_tap_route_ids.len() != rtp_tap_payload_types.len() {
333            return Err(JsValue::from_str(
334                "RTP tap route id and payload type arrays must have the same length",
335            ));
336        }
337        let rtp_payload_taps = rtp_tap_route_ids
338            .iter()
339            .zip(rtp_tap_payload_types.iter())
340            .map(|(route_id, payload_type)| RtpPayloadTap {
341                route_id: PayloadRouteId::new(*route_id as u64),
342                payload_type: *payload_type,
343            })
344            .collect::<Vec<_>>();
345        self.push_rx_transfer_profiled_inner(
346            transfer,
347            keep_corrupted,
348            raw_route_ids,
349            &rtp_payload_taps,
350        )
351    }
352
353    #[wasm_bindgen(js_name = addKeyedRoute)]
354    /// Add an encrypted raw-payload route to the receiver.
355    pub fn add_keyed_route(
356        &mut self,
357        route_id: u32,
358        channel_id: u32,
359        keypair: &[u8],
360        minimum_epoch: u64,
361    ) -> Result<(), JsValue> {
362        let keypair = WfbKeypair::from_bytes(keypair)
363            .map_err(|err| JsValue::from_str(&format!("invalid WFB keypair: {err}")))?;
364        self.runtime
365            .add_keyed_route(
366                PayloadRouteId::new(route_id as u64),
367                ChannelId::new(channel_id),
368                DEFAULT_KEY_SLOT,
369                keypair,
370                minimum_epoch,
371            )
372            .map_err(|err| JsValue::from_str(&format!("invalid route config: {err}")))?;
373        Ok(())
374    }
375
376    #[wasm_bindgen(js_name = fecCounters)]
377    /// Return cumulative video FEC counters as JSON.
378    pub fn fec_counters(&self) -> String {
379        counters_json(self.video_fec_counters())
380    }
381}
382
383impl OpenIpcReceiver {
384    fn push_rx_transfer_profiled_inner(
385        &mut self,
386        transfer: &[u8],
387        keep_corrupted: bool,
388        raw_route_ids: &[u32],
389        rtp_payload_taps: &[RtpPayloadTap],
390    ) -> Result<Object, JsValue> {
391        let total_start = now_ms();
392        let parse_start = now_ms();
393        let packets = parse_rx_aggregate_with_kind(transfer, self.rx_descriptor_kind)
394            .map_err(|err| JsValue::from_str(&format!("Realtek RX aggregate rejected: {err}")))?;
395        let parse_ms = elapsed_ms(parse_start);
396
397        let raw_payload_routes = raw_route_ids
398            .iter()
399            .map(|id| PayloadRouteId::new(*id as u64))
400            .collect();
401        let pipeline_start = now_ms();
402        let batch = self.runtime.push_rx_packets(
403            packets,
404            &ReceiverBatchOptions {
405                accept_corrupted: keep_corrupted,
406                raw_payload_routes,
407                rtp_payload_taps: rtp_payload_taps.to_vec(),
408                depacketize_video: true,
409            },
410        );
411        let pipeline_ms = elapsed_ms(pipeline_start);
412        receiver_profile_object(
413            batch,
414            transfer.len(),
415            parse_ms,
416            pipeline_ms,
417            elapsed_ms(total_start),
418        )
419    }
420}
421
422pub(crate) fn receiver_profile_object(
423    batch: ReceiverBatch,
424    transfer_bytes: usize,
425    parse_ms: f64,
426    pipeline_ms: f64,
427    total_ms: f64,
428) -> Result<Object, JsValue> {
429    let counters = batch.counters;
430    let fec_counters = batch.fec_counters;
431    let frames = frame_objects_array(batch.frames)?;
432    let raw_payloads = raw_payload_array(batch.raw_payloads)?;
433    let rtp_status = rtp_status_object(batch.rtp_status, batch.rtp_reorder_status)?;
434
435    let object = Object::new();
436    Reflect::set(&object, &JsValue::from_str("frames"), &frames)?;
437    Reflect::set(&object, &JsValue::from_str("rawPayloads"), &raw_payloads)?;
438    Reflect::set(
439        &object,
440        &JsValue::from_str("mavlinkPayloads"),
441        &raw_payloads,
442    )?;
443    Reflect::set(&object, &JsValue::from_str("rtpStatus"), &rtp_status)?;
444    Reflect::set(
445        &object,
446        &JsValue::from_str("fecCounters"),
447        &counters_object(fec_counters)?.into(),
448    )?;
449    set_number(
450        &object,
451        "rawPayloadCount",
452        counters.raw_payload_count as f64,
453    )?;
454    set_number(
455        &object,
456        "rawPayloadBytes",
457        counters.raw_payload_bytes as f64,
458    )?;
459    set_number(&object, "transferBytes", transfer_bytes as f64)?;
460    set_number(&object, "packets", counters.packets as f64)?;
461    set_number(&object, "acceptedPackets", counters.accepted_packets as f64)?;
462    set_number(&object, "droppedPackets", counters.dropped_packets as f64)?;
463    set_number(&object, "crcDropped", counters.crc_dropped as f64)?;
464    set_number(&object, "icvDropped", counters.icv_dropped as f64)?;
465    set_number(&object, "reportDropped", counters.report_dropped as f64)?;
466    set_number(&object, "ignoredFrames", counters.ignored_frames as f64)?;
467    set_number(&object, "sessions", counters.sessions as f64)?;
468    set_number(&object, "wfbPayloads", counters.wfb_payloads as f64)?;
469    set_number(&object, "rtpPackets", counters.rtp_packets as f64)?;
470    set_number(&object, "videoFrames", counters.video_frames as f64)?;
471    set_number(
472        &object,
473        "mavlinkPayloadCount",
474        counters.raw_payload_count as f64,
475    )?;
476    set_number(&object, "mavlinkBytes", counters.raw_payload_bytes as f64)?;
477    set_number(&object, "parseMs", parse_ms)?;
478    set_number(&object, "pipelineMs", pipeline_ms)?;
479    set_number(&object, "totalMs", total_ms)?;
480    Ok(object)
481}
482
483fn openipc_receiver_with_keypair_and_telemetry_channel_inner(
484    channel_id: u32,
485    telemetry_channel_id: u32,
486    keypair: WfbKeypair,
487    minimum_epoch: u64,
488) -> Result<OpenIpcReceiver, JsValue> {
489    let mut runtime = ReceiverRuntime::with_keyed_video_route(
490        FrameLayout::WithFcs,
491        VIDEO_ROUTE_ID,
492        ChannelId::new(channel_id),
493        DEFAULT_KEY_SLOT,
494        keypair,
495        minimum_epoch,
496    )
497    .map_err(|err| JsValue::from_str(&format!("invalid encrypted receiver config: {err}")))?;
498    runtime
499        .add_keyed_route(
500            TELEMETRY_ROUTE_ID,
501            ChannelId::new(telemetry_channel_id),
502            DEFAULT_KEY_SLOT,
503            keypair,
504            minimum_epoch,
505        )
506        .map_err(|err| JsValue::from_str(&format!("invalid MAVLink receiver config: {err}")))?;
507    Ok(OpenIpcReceiver {
508        runtime,
509        rx_descriptor_kind: RxDescriptorKind::Jaguar1,
510    })
511}
512
513pub(crate) fn parse_rx_descriptor_kind(kind: &str) -> Result<RxDescriptorKind, JsValue> {
514    match kind {
515        "jaguar1" | "rtl8812" | "rtl8821" | "rtl8814" => Ok(RxDescriptorKind::Jaguar1),
516        "jaguar3" | "rtl8812cu" | "rtl8812eu" | "rtl8822c" | "rtl8822cu" | "rtl8822e"
517        | "rtl8822eu" => Ok(RxDescriptorKind::Jaguar3),
518        _ => Err(JsValue::from_str(
519            "unsupported RX descriptor kind; expected jaguar1 or jaguar3",
520        )),
521    }
522}
523
524fn frame_bytes_array(frames: Vec<openipc_core::DepacketizedFrame>) -> Array {
525    let out = Array::new();
526    for frame in frames {
527        out.push(&Uint8Array::from(frame.data.as_slice()));
528    }
529    out
530}
531
532pub(crate) fn frame_objects_array(
533    frames: Vec<openipc_core::DepacketizedFrame>,
534) -> Result<Array, JsValue> {
535    let out = Array::new();
536    for frame in frames {
537        out.push(&video_frame_object(frame)?.into());
538    }
539    Ok(out)
540}
541
542pub(crate) fn raw_payload_array(
543    payloads: Vec<openipc_core::RoutePayload>,
544) -> Result<Array, JsValue> {
545    let out = Array::new();
546    for payload in payloads {
547        out.push(&raw_payload_object(payload)?.into());
548    }
549    Ok(out)
550}
openipc_web/receiver.rs

openipc_web/
receiver.rs
