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

openipc_web/
receiver.rs
