Skip to main content

rns_net/interface/rnode/
mod.rs

1//! RNode LoRa radio interface.
2//!
3//! Manages serial connection to RNode device, detects firmware,
4//! configures radio parameters per subinterface. Supports single-radio
5//! (RNodeInterface) and multi-radio (RNodeMultiInterface) devices.
6//!
7//! Matches Python `RNodeInterface.py` and `RNodeMultiInterface.py`.
8
9mod transport;
10
11use std::io::{self, Read, Write};
12use std::sync::{Arc, Mutex};
13use std::thread;
14use std::time::Duration;
15use transport::Transport;
16
17use rns_core::transport::types::{AirtimeProfile, InterfaceId};
18
19use crate::event::{Event, EventSender};
20use crate::interface::{lock_or_recover, Writer};
21use crate::rnode_kiss;
22use crate::serial::{Parity, SerialConfig};
23
24/// Validation limits matching Python RNodeInterface.
25pub const FREQ_MIN: u32 = 137_000_000;
26pub const FREQ_MAX: u32 = 3_000_000_000;
27pub const BW_MIN: u32 = 7_800;
28pub const BW_MAX: u32 = 1_625_000;
29pub const SF_MIN: u8 = 5;
30pub const SF_MAX: u8 = 12;
31pub const CR_MIN: u8 = 5;
32pub const CR_MAX: u8 = 8;
33pub const TXPOWER_MIN: i8 = 0;
34pub const TXPOWER_MAX: i8 = 37;
35pub const HW_MTU: u32 = 508;
36pub const LORA_PREAMBLE_SYMBOLS: u16 = 8;
37pub const LORA_EXPLICIT_HEADER: bool = true;
38pub const LORA_CRC: bool = true;
39
40/// Configuration for one RNode subinterface (radio).
41#[derive(Debug, Clone)]
42pub struct RNodeSubConfig {
43    pub name: String,
44    pub vport: u8,
45    pub outgoing: bool,
46    pub frequency: u32,
47    pub bandwidth: u32,
48    pub txpower: i8,
49    pub spreading_factor: u8,
50    pub coding_rate: u8,
51    pub flow_control: bool,
52    pub st_alock: Option<f32>,
53    pub lt_alock: Option<f32>,
54}
55
56/// Configuration for the RNode device.
57#[derive(Debug, Clone)]
58pub struct RNodeConfig {
59    pub name: String,
60    pub port: String,
61    pub speed: u32,
62    pub subinterfaces: Vec<RNodeSubConfig>,
63    pub multi: bool,
64    pub id_interval: Option<u32>,
65    pub id_callsign: Option<Vec<u8>>,
66    pub base_interface_id: InterfaceId,
67    /// Pre-opened file descriptor (e.g. from a USB bridge socketpair on Android).
68    /// When set, `start()` uses this fd directly instead of opening `port`.
69    pub pre_opened_fd: Option<i32>,
70    pub underlay_mark: Option<u32>,
71    pub runtime: Arc<Mutex<RNodeRuntime>>,
72}
73
74#[derive(Debug, Clone)]
75pub struct RNodeRuntime {
76    pub sub: RNodeSubConfig,
77    pub writer: Option<Arc<Mutex<Transport>>>,
78}
79
80impl RNodeRuntime {
81    pub fn from_config(config: &RNodeConfig) -> Self {
82        Self {
83            sub: config
84                .subinterfaces
85                .first()
86                .cloned()
87                .unwrap_or_else(|| RNodeSubConfig {
88                    name: config.name.clone(),
89                    vport: 0,
90                    outgoing: true,
91                    frequency: 868_000_000,
92                    bandwidth: 125_000,
93                    txpower: 7,
94                    spreading_factor: 8,
95                    coding_rate: 5,
96                    flow_control: false,
97                    st_alock: None,
98                    lt_alock: None,
99                }),
100            writer: None,
101        }
102    }
103}
104
105#[derive(Debug, Clone)]
106pub struct RNodeRuntimeConfigHandle {
107    pub interface_name: String,
108    pub runtime: Arc<Mutex<RNodeRuntime>>,
109    pub startup: RNodeRuntime,
110}
111
112impl Default for RNodeConfig {
113    fn default() -> Self {
114        let mut config = RNodeConfig {
115            name: String::new(),
116            port: String::new(),
117            speed: 115200,
118            subinterfaces: Vec::new(),
119            multi: false,
120            id_interval: None,
121            id_callsign: None,
122            base_interface_id: InterfaceId(0),
123            pre_opened_fd: None,
124            underlay_mark: None,
125            runtime: Arc::new(Mutex::new(RNodeRuntime {
126                sub: RNodeSubConfig {
127                    name: String::new(),
128                    vport: 0,
129                    outgoing: true,
130                    frequency: 868_000_000,
131                    bandwidth: 125_000,
132                    txpower: 7,
133                    spreading_factor: 8,
134                    coding_rate: 5,
135                    flow_control: false,
136                    st_alock: None,
137                    lt_alock: None,
138                },
139                writer: None,
140            })),
141        };
142        let startup = RNodeRuntime::from_config(&config);
143        config.runtime = Arc::new(Mutex::new(startup));
144        config
145    }
146}
147
148/// Validate subinterface configuration. Returns error message if invalid.
149pub fn validate_sub_config(sub: &RNodeSubConfig) -> Option<String> {
150    if sub.frequency < FREQ_MIN || sub.frequency > FREQ_MAX {
151        return Some(format!(
152            "Invalid frequency {} for {}",
153            sub.frequency, sub.name
154        ));
155    }
156    if sub.bandwidth < BW_MIN || sub.bandwidth > BW_MAX {
157        return Some(format!(
158            "Invalid bandwidth {} for {}",
159            sub.bandwidth, sub.name
160        ));
161    }
162    if sub.spreading_factor < SF_MIN || sub.spreading_factor > SF_MAX {
163        return Some(format!(
164            "Invalid SF {} for {}",
165            sub.spreading_factor, sub.name
166        ));
167    }
168    if sub.coding_rate < CR_MIN || sub.coding_rate > CR_MAX {
169        return Some(format!("Invalid CR {} for {}", sub.coding_rate, sub.name));
170    }
171    if sub.txpower < TXPOWER_MIN || sub.txpower > TXPOWER_MAX {
172        return Some(format!("Invalid TX power {} for {}", sub.txpower, sub.name));
173    }
174    if let Some(st) = sub.st_alock {
175        if !(0.0..=100.0).contains(&st) {
176            return Some(format!("Invalid ST airtime limit {} for {}", st, sub.name));
177        }
178    }
179    if let Some(lt) = sub.lt_alock {
180        if !(0.0..=100.0).contains(&lt) {
181            return Some(format!("Invalid LT airtime limit {} for {}", lt, sub.name));
182        }
183    }
184    None
185}
186
187/// Estimate the LoRa physical bit rate for announce bandwidth gating.
188///
189/// Formula: SF * (bandwidth / 2^SF) * (4 / coding_rate).
190pub fn estimate_lora_bitrate_bps(sub: &RNodeSubConfig) -> u64 {
191    let symbols_per_second = sub.bandwidth as f64 / (1u64 << sub.spreading_factor) as f64;
192    let bits_per_symbol = sub.spreading_factor as f64 * (4.0 / sub.coding_rate as f64);
193    (symbols_per_second * bits_per_symbol).round().max(1.0) as u64
194}
195
196pub fn lora_airtime_profile(sub: &RNodeSubConfig) -> AirtimeProfile {
197    AirtimeProfile::Lora {
198        bandwidth: sub.bandwidth,
199        spreading_factor: sub.spreading_factor,
200        coding_rate: sub.coding_rate,
201        preamble_symbols: LORA_PREAMBLE_SYMBOLS,
202        explicit_header: LORA_EXPLICIT_HEADER,
203        crc: LORA_CRC,
204    }
205}
206
207/// Writer for a specific RNode subinterface.
208/// Wraps a shared serial writer with subinterface-specific data framing.
209struct RNodeSubWriter {
210    writer: Arc<Mutex<Transport>>,
211    index: u8,
212    flow_control: bool,
213    flow_state: Arc<Mutex<SubFlowState>>,
214    multi: bool,
215}
216
217struct SubFlowState {
218    ready: bool,
219    queue: std::collections::VecDeque<Vec<u8>>,
220}
221
222fn make_sub_writer(
223    writer: Arc<Mutex<Transport>>,
224    index: u8,
225    flow_control: bool,
226    flow_state: Arc<Mutex<SubFlowState>>,
227    multi: bool,
228) -> Box<dyn Writer> {
229    Box::new(RNodeSubWriter {
230        writer,
231        index,
232        flow_control,
233        flow_state,
234        multi,
235    })
236}
237
238impl Writer for RNodeSubWriter {
239    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
240        let frame = if self.multi {
241            rnode_kiss::rnode_multi_data_frame(self.index, data)
242        } else {
243            rnode_kiss::rnode_data_frame(self.index, data)
244        };
245        if self.flow_control {
246            let mut state = lock_or_recover(&self.flow_state, "rnode flow state");
247            if state.ready {
248                state.ready = false;
249                drop(state);
250                lock_or_recover(&self.writer, "rnode shared writer").write_all(&frame)
251            } else {
252                state.queue.push_back(data.to_vec());
253                Ok(())
254            }
255        } else {
256            lock_or_recover(&self.writer, "rnode shared writer").write_all(&frame)
257        }
258    }
259}
260
261/// Start the RNode interface.
262///
263/// Opens serial port, spawns reader thread which performs detect+configure,
264/// then enters data relay mode.
265///
266/// Returns one `(InterfaceId, Box<dyn Writer>)` per subinterface.
267pub fn start(
268    config: RNodeConfig,
269    tx: EventSender,
270) -> io::Result<Vec<(InterfaceId, Box<dyn Writer>)>> {
271    // Validate all subinterface configs upfront
272    for sub in &config.subinterfaces {
273        if let Some(err) = validate_sub_config(sub) {
274            return Err(io::Error::new(io::ErrorKind::InvalidInput, err));
275        }
276    }
277
278    if config.subinterfaces.is_empty() {
279        return Err(io::Error::new(
280            io::ErrorKind::InvalidInput,
281            "No subinterfaces configured",
282        ));
283    }
284
285    let (reader_file, shared_writer) = if let Some(fd) = config.pre_opened_fd {
286        // Pre-opened fd from USB bridge — dup for independent reader/writer handles
287        let (r, w) = Transport::open_from_fd(fd)?;
288        (r, Arc::new(Mutex::new(w)))
289    } else {
290        let serial_config = SerialConfig {
291            path: config.port.clone(),
292            baud: config.speed,
293            data_bits: 8,
294            parity: Parity::None,
295            stop_bits: 1,
296        };
297        let (r, w) = Transport::open(&serial_config, config.underlay_mark)?;
298        (r, Arc::new(Mutex::new(w)))
299    };
300
301    // Build per-subinterface writers and IDs
302    let num_subs = config.subinterfaces.len();
303    let mut writers: Vec<(InterfaceId, Box<dyn Writer>)> = Vec::with_capacity(num_subs);
304    let mut flow_states: Vec<Arc<Mutex<SubFlowState>>> = Vec::with_capacity(num_subs);
305
306    for (i, sub) in config.subinterfaces.iter().enumerate() {
307        let sub_id = InterfaceId(config.base_interface_id.0 + i as u64);
308        let flow_state = Arc::new(Mutex::new(SubFlowState {
309            ready: true,
310            queue: std::collections::VecDeque::new(),
311        }));
312        flow_states.push(flow_state.clone());
313        let sub_writer = make_sub_writer(
314            shared_writer.clone(),
315            sub.vport,
316            sub.flow_control,
317            flow_state,
318            config.multi,
319        );
320        writers.push((sub_id, sub_writer));
321    }
322
323    // Spawn reader thread
324    let reader_shared_writer = shared_writer.clone();
325    {
326        let mut runtime = lock_or_recover(&config.runtime, "rnode runtime");
327        runtime.writer = Some(shared_writer.clone());
328        runtime.sub = config
329            .subinterfaces
330            .first()
331            .cloned()
332            .unwrap_or(runtime.sub.clone());
333    }
334    let reader_config = config.clone();
335    let reader_flow_states = flow_states;
336    thread::Builder::new()
337        .name(format!("rnode-reader-{}", config.base_interface_id.0))
338        .spawn(move || {
339            reader_loop(
340                reader_file,
341                reader_shared_writer,
342                reader_config,
343                tx,
344                reader_flow_states,
345            );
346        })?;
347
348    // Spawn keepalive thread — sends periodic DETECT to prevent firmware
349    // bridge idle timeout (ESP32 RNode reverts to standalone after 30s idle).
350    let keepalive_writer = shared_writer.clone();
351    let keepalive_name = config.name.clone();
352    thread::Builder::new()
353        .name(format!("rnode-keepalive-{}", config.base_interface_id.0))
354        .spawn(move || {
355            let detect = rnode_kiss::detect_request();
356            loop {
357                thread::sleep(Duration::from_secs(15));
358                if let Err(e) =
359                    lock_or_recover(&keepalive_writer, "rnode shared writer").write_all(&detect)
360                {
361                    log::debug!("[{}] keepalive write failed: {}", keepalive_name, e);
362                }
363            }
364        })?;
365
366    Ok(writers)
367}
368
369/// Reader loop: detect device, configure radios, then relay data frames.
370fn reader_loop(
371    mut reader: Transport,
372    writer: Arc<Mutex<Transport>>,
373    config: RNodeConfig,
374    tx: EventSender,
375    flow_states: Vec<Arc<Mutex<SubFlowState>>>,
376) {
377    const RECONNECT_INITIAL_DELAY: Duration = Duration::from_millis(200);
378    const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(2);
379    // Initial delay for hardware init (matches Python: sleep(2.0))
380    thread::sleep(Duration::from_secs(2));
381    let mut connected_once = false;
382    // RNode is the PHY-stat-capable interface: retain each modem report until
383    // the matching frame is enqueued, attach it to that frame, then clear it
384    // so later frames cannot inherit stale RSSI or SNR.
385    let mut last_rssi: Option<i16> = None;
386    let mut last_snr: Option<f32> = None;
387    if let Err(e) = detect_and_configure(&mut reader, &writer, &config) {
388        log::error!("[{}] initial RNode setup failed: {}", config.name, e);
389        return;
390    }
391    signal_interface_up(&tx, &config, &writer, &flow_states, connected_once);
392    connected_once = true;
393    loop {
394        let mut decoder = rnode_kiss::RNodeDecoder::new();
395        let mut buf = [0u8; 4096];
396        let disconnected = loop {
397            match reader.read(&mut buf) {
398                Ok(0) => {
399                    log::warn!("[{}] serial port closed", config.name);
400                    signal_interface_down(&tx, &config);
401                    break true;
402                }
403                Ok(n) => {
404                    for event in decoder.feed(&buf[..n]) {
405                        if update_pending_rx_metadata(&event, &mut last_rssi, &mut last_snr) {
406                            continue;
407                        }
408                        match event {
409                            rnode_kiss::RNodeEvent::DataFrame { index, data } => {
410                                let Some(position) = config
411                                    .subinterfaces
412                                    .iter()
413                                    .position(|sub| usize::from(sub.vport) == index)
414                                else {
415                                    continue;
416                                };
417                                let sub_id =
418                                    InterfaceId(config.base_interface_id.0 + position as u64);
419                                if tx
420                                    .send(Event::Frame {
421                                        interface_id: sub_id,
422                                        data,
423                                        rssi: last_rssi,
424                                        snr: last_snr,
425                                    })
426                                    .is_err()
427                                {
428                                    return;
429                                }
430                                last_rssi = None;
431                                last_snr = None;
432                            }
433                            rnode_kiss::RNodeEvent::Ready => {
434                                // Flow control: unlock all subinterfaces that have flow_control
435                                for (i, fs) in flow_states.iter().enumerate() {
436                                    if config.subinterfaces[i].flow_control {
437                                        process_flow_queue(
438                                            fs,
439                                            &writer,
440                                            config.subinterfaces[i].vport,
441                                            config.multi,
442                                        );
443                                    }
444                                }
445                            }
446                            rnode_kiss::RNodeEvent::Error(code) => {
447                                log::error!("[{}] RNode error: 0x{:02X}", config.name, code);
448                            }
449                            _ => {
450                                // Status updates logged but not acted on
451                            }
452                        }
453                    }
454                }
455                Err(e) => {
456                    log::error!("[{}] serial read error: {}", config.name, e);
457                    signal_interface_down(&tx, &config);
458                    break true;
459                }
460            }
461        };
462
463        clear_pending_rx_metadata(&mut last_rssi, &mut last_snr);
464
465        if !disconnected || config.pre_opened_fd.is_some() {
466            return;
467        }
468
469        let mut backoff = RECONNECT_INITIAL_DELAY;
470        loop {
471            match reopen_connection(&config, &writer) {
472                Ok(new_reader) => {
473                    reset_flow_states(&flow_states);
474                    reader = new_reader;
475                    if let Err(e) = detect_and_configure(&mut reader, &writer, &config) {
476                        log::warn!("[{}] reconnect configure failed: {}", config.name, e);
477                        thread::sleep(backoff);
478                        backoff = std::cmp::min(backoff.saturating_mul(2), RECONNECT_MAX_DELAY);
479                        continue;
480                    }
481                    signal_interface_up(&tx, &config, &writer, &flow_states, connected_once);
482                    break;
483                }
484                Err(e) => {
485                    log::warn!("[{}] reconnect open failed: {}", config.name, e);
486                    thread::sleep(backoff);
487                    backoff = std::cmp::min(backoff.saturating_mul(2), RECONNECT_MAX_DELAY);
488                }
489            }
490        }
491    }
492}
493
494fn detect_and_configure(
495    reader: &mut Transport,
496    writer: &Arc<Mutex<Transport>>,
497    config: &RNodeConfig,
498) -> io::Result<()> {
499    let detect_cmd = rnode_kiss::detect_request();
500    let mut cmd = detect_cmd;
501    cmd.extend_from_slice(&rnode_kiss::rnode_command(
502        rnode_kiss::CMD_FW_VERSION,
503        &[0x00],
504    ));
505    cmd.extend_from_slice(&rnode_kiss::rnode_command(
506        rnode_kiss::CMD_FW_DETAIL,
507        &[0x00],
508    ));
509    cmd.extend_from_slice(&rnode_kiss::rnode_command(
510        rnode_kiss::CMD_PLATFORM,
511        &[0x00],
512    ));
513    cmd.extend_from_slice(&rnode_kiss::rnode_command(rnode_kiss::CMD_MCU, &[0x00]));
514
515    lock_or_recover(writer, "rnode shared writer").write_all(&cmd)?;
516
517    let mut decoder = rnode_kiss::RNodeDecoder::new();
518    let mut buf = [0u8; 4096];
519    let mut detected = false;
520    let detect_start = std::time::Instant::now();
521    let detect_timeout = Duration::from_secs(5);
522
523    while !detected && detect_start.elapsed() < detect_timeout {
524        match reader.read(&mut buf) {
525            Ok(0) => {
526                return Err(io::Error::new(
527                    io::ErrorKind::UnexpectedEof,
528                    "serial port closed during detect",
529                ));
530            }
531            Ok(n) => {
532                for event in decoder.feed(&buf[..n]) {
533                    match event {
534                        rnode_kiss::RNodeEvent::Detected(true) => {
535                            detected = true;
536                            log::info!("[{}] RNode device detected", config.name);
537                        }
538                        rnode_kiss::RNodeEvent::FirmwareVersion { major, minor } => {
539                            log::info!("[{}] firmware version {}.{}", config.name, major, minor);
540                        }
541                        rnode_kiss::RNodeEvent::FirmwareDetail(ref detail) => {
542                            log::info!("[{}] firmware detail: {}", config.name, detail);
543                        }
544                        rnode_kiss::RNodeEvent::Platform(p) => {
545                            log::info!("[{}] platform: 0x{:02X}", config.name, p);
546                        }
547                        rnode_kiss::RNodeEvent::Mcu(m) => {
548                            log::info!("[{}] MCU: 0x{:02X}", config.name, m);
549                        }
550                        _ => {}
551                    }
552                }
553            }
554            Err(e) => {
555                return Err(io::Error::new(
556                    e.kind(),
557                    format!("serial read error during detect: {}", e),
558                ));
559            }
560        }
561    }
562
563    if !detected {
564        return Err(io::Error::new(
565            io::ErrorKind::TimedOut,
566            "RNode detection timed out",
567        ));
568    }
569
570    for sub in &config.subinterfaces {
571        configure_subinterface(writer, sub.vport, sub, config.multi)?;
572    }
573
574    thread::sleep(Duration::from_millis(300));
575    log::info!(
576        "[{}] RNode configured with {} subinterface(s)",
577        config.name,
578        config.subinterfaces.len()
579    );
580    Ok(())
581}
582
583fn signal_interface_down(tx: &EventSender, config: &RNodeConfig) {
584    for i in 0..config.subinterfaces.len() {
585        let sub_id = InterfaceId(config.base_interface_id.0 + i as u64);
586        let _ = tx.send(Event::InterfaceDown(sub_id));
587    }
588}
589
590fn update_pending_rx_metadata(
591    event: &rnode_kiss::RNodeEvent,
592    last_rssi: &mut Option<i16>,
593    last_snr: &mut Option<f32>,
594) -> bool {
595    match event {
596        rnode_kiss::RNodeEvent::StatRssi(rssi) => {
597            *last_rssi = Some(*rssi as i16 - 157);
598            true
599        }
600        rnode_kiss::RNodeEvent::StatSnr(snr) => {
601            *last_snr = Some(*snr as f32 * 0.25);
602            true
603        }
604        _ => false,
605    }
606}
607
608fn signal_interface_up(
609    tx: &EventSender,
610    config: &RNodeConfig,
611    writer: &Arc<Mutex<Transport>>,
612    flow_states: &[Arc<Mutex<SubFlowState>>],
613    reconnected: bool,
614) {
615    for (i, flow_state) in flow_states.iter().enumerate() {
616        let sub_id = InterfaceId(config.base_interface_id.0 + i as u64);
617        let new_writer = reconnected.then(|| {
618            make_sub_writer(
619                writer.clone(),
620                config.subinterfaces[i].vport,
621                config.subinterfaces[i].flow_control,
622                flow_state.clone(),
623                config.multi,
624            )
625        });
626        let _ = tx.send(Event::InterfaceUp(sub_id, new_writer, None));
627    }
628}
629
630fn reset_flow_states(flow_states: &[Arc<Mutex<SubFlowState>>]) {
631    for flow_state in flow_states {
632        let mut state = lock_or_recover(flow_state, "rnode flow state");
633        state.ready = true;
634        state.queue.clear();
635    }
636}
637
638fn clear_pending_rx_metadata(last_rssi: &mut Option<i16>, last_snr: &mut Option<f32>) {
639    *last_rssi = None;
640    *last_snr = None;
641}
642
643fn reopen_connection(
644    config: &RNodeConfig,
645    writer: &Arc<Mutex<Transport>>,
646) -> io::Result<Transport> {
647    let serial_config = SerialConfig {
648        path: config.port.clone(),
649        baud: config.speed,
650        data_bits: 8,
651        parity: Parity::None,
652        stop_bits: 1,
653    };
654
655    let (reader, new_writer) = Transport::open(&serial_config, config.underlay_mark)?;
656    *lock_or_recover(writer, "rnode shared writer") = new_writer;
657    Ok(reader)
658}
659
660/// Configure a single subinterface on the RNode device.
661pub(crate) fn configure_subinterface(
662    writer: &Arc<Mutex<Transport>>,
663    index: u8,
664    sub: &RNodeSubConfig,
665    multi: bool,
666) -> io::Result<()> {
667    let mut w = lock_or_recover(writer, "rnode shared writer");
668
669    // For multi-radio, send select command before each parameter
670    let freq_bytes = [
671        (sub.frequency >> 24) as u8,
672        (sub.frequency >> 16) as u8,
673        (sub.frequency >> 8) as u8,
674        (sub.frequency & 0xFF) as u8,
675    ];
676    let bw_bytes = [
677        (sub.bandwidth >> 24) as u8,
678        (sub.bandwidth >> 16) as u8,
679        (sub.bandwidth >> 8) as u8,
680        (sub.bandwidth & 0xFF) as u8,
681    ];
682    let txp = if sub.txpower < 0 {
683        (sub.txpower as i16 + 256) as u8
684    } else {
685        sub.txpower as u8
686    };
687
688    if multi {
689        w.write_all(&rnode_kiss::rnode_select_command(
690            index,
691            rnode_kiss::CMD_FREQUENCY,
692            &freq_bytes,
693        ))?;
694        w.write_all(&rnode_kiss::rnode_select_command(
695            index,
696            rnode_kiss::CMD_BANDWIDTH,
697            &bw_bytes,
698        ))?;
699        w.write_all(&rnode_kiss::rnode_select_command(
700            index,
701            rnode_kiss::CMD_TXPOWER,
702            &[txp],
703        ))?;
704        w.write_all(&rnode_kiss::rnode_select_command(
705            index,
706            rnode_kiss::CMD_SF,
707            &[sub.spreading_factor],
708        ))?;
709        w.write_all(&rnode_kiss::rnode_select_command(
710            index,
711            rnode_kiss::CMD_CR,
712            &[sub.coding_rate],
713        ))?;
714    } else {
715        w.write_all(&rnode_kiss::rnode_command(
716            rnode_kiss::CMD_FREQUENCY,
717            &freq_bytes,
718        ))?;
719        w.write_all(&rnode_kiss::rnode_command(
720            rnode_kiss::CMD_BANDWIDTH,
721            &bw_bytes,
722        ))?;
723        w.write_all(&rnode_kiss::rnode_command(rnode_kiss::CMD_TXPOWER, &[txp]))?;
724        w.write_all(&rnode_kiss::rnode_command(
725            rnode_kiss::CMD_SF,
726            &[sub.spreading_factor],
727        ))?;
728        w.write_all(&rnode_kiss::rnode_command(
729            rnode_kiss::CMD_CR,
730            &[sub.coding_rate],
731        ))?;
732    }
733
734    // Airtime locks
735    if let Some(st) = sub.st_alock {
736        let st_val = (st * 100.0) as u16;
737        let st_bytes = [(st_val >> 8) as u8, (st_val & 0xFF) as u8];
738        if multi {
739            w.write_all(&rnode_kiss::rnode_select_command(
740                index,
741                rnode_kiss::CMD_ST_ALOCK,
742                &st_bytes,
743            ))?;
744        } else {
745            w.write_all(&rnode_kiss::rnode_command(
746                rnode_kiss::CMD_ST_ALOCK,
747                &st_bytes,
748            ))?;
749        }
750    }
751    if let Some(lt) = sub.lt_alock {
752        let lt_val = (lt * 100.0) as u16;
753        let lt_bytes = [(lt_val >> 8) as u8, (lt_val & 0xFF) as u8];
754        if multi {
755            w.write_all(&rnode_kiss::rnode_select_command(
756                index,
757                rnode_kiss::CMD_LT_ALOCK,
758                &lt_bytes,
759            ))?;
760        } else {
761            w.write_all(&rnode_kiss::rnode_command(
762                rnode_kiss::CMD_LT_ALOCK,
763                &lt_bytes,
764            ))?;
765        }
766    }
767
768    // Turn on radio
769    if multi {
770        w.write_all(&rnode_kiss::rnode_select_command(
771            index,
772            rnode_kiss::CMD_RADIO_STATE,
773            &[rnode_kiss::RADIO_STATE_ON],
774        ))?;
775    } else {
776        w.write_all(&rnode_kiss::rnode_command(
777            rnode_kiss::CMD_RADIO_STATE,
778            &[rnode_kiss::RADIO_STATE_ON],
779        ))?;
780    }
781
782    Ok(())
783}
784
785/// Process flow control queue for a subinterface.
786fn process_flow_queue(
787    flow_state: &Arc<Mutex<SubFlowState>>,
788    writer: &Arc<Mutex<Transport>>,
789    index: u8,
790    multi: bool,
791) {
792    let mut state = lock_or_recover(flow_state, "rnode flow state");
793    if let Some(data) = state.queue.pop_front() {
794        state.ready = false;
795        drop(state);
796        let frame = if multi {
797            rnode_kiss::rnode_multi_data_frame(index, &data)
798        } else {
799            rnode_kiss::rnode_data_frame(index, &data)
800        };
801        let _ = lock_or_recover(writer, "rnode shared writer").write_all(&frame);
802    } else {
803        state.ready = true;
804    }
805}
806
807// --- Factory implementation ---
808
809use super::{
810    ConfigSection, InterfaceConfigData, InterfaceFactory, StartContext, StartResult, SubInterface,
811};
812use rns_core::transport::types::InterfaceInfo;
813use std::collections::HashMap;
814
815/// Factory for `RNodeInterface`.
816pub struct RNodeFactory;
817
818impl InterfaceFactory for RNodeFactory {
819    fn type_name(&self) -> &str {
820        "RNodeInterface"
821    }
822
823    fn default_ifac_size(&self) -> usize {
824        8
825    }
826
827    fn parse_config(
828        &self,
829        name: &str,
830        id: InterfaceId,
831        params: &HashMap<String, String>,
832    ) -> Result<Box<dyn InterfaceConfigData>, String> {
833        let pre_opened_fd = params.get("fd").and_then(|v| v.parse::<i32>().ok());
834
835        let port = params
836            .get("port")
837            .cloned()
838            .or_else(|| pre_opened_fd.map(|_| "usb-bridge".to_string()))
839            .ok_or_else(|| "RNodeInterface requires 'port' or 'fd'".to_string())?;
840
841        let speed = params
842            .get("speed")
843            .and_then(|v| v.parse().ok())
844            .unwrap_or(115200u32);
845
846        let frequency = params
847            .get("frequency")
848            .and_then(|v| v.parse().ok())
849            .unwrap_or(868_000_000u32);
850
851        let bandwidth = params
852            .get("bandwidth")
853            .and_then(|v| v.parse().ok())
854            .unwrap_or(125_000u32);
855
856        let txpower = params
857            .get("txpower")
858            .and_then(|v| v.parse().ok())
859            .unwrap_or(7i8);
860
861        let spreading_factor = params
862            .get("spreadingfactor")
863            .or_else(|| params.get("spreading_factor"))
864            .and_then(|v| v.parse().ok())
865            .unwrap_or(8u8);
866
867        let coding_rate = params
868            .get("codingrate")
869            .or_else(|| params.get("coding_rate"))
870            .and_then(|v| v.parse().ok())
871            .unwrap_or(5u8);
872
873        let flow_control = params
874            .get("flow_control")
875            .and_then(|v| crate::config::parse_bool_pub(v))
876            .unwrap_or(false);
877
878        let st_alock = params.get("st_alock").and_then(|v| v.parse().ok());
879
880        let lt_alock = params.get("lt_alock").and_then(|v| v.parse().ok());
881
882        let id_interval = params.get("id_interval").and_then(|v| v.parse().ok());
883
884        let id_callsign = params.get("id_callsign").map(|v| v.as_bytes().to_vec());
885
886        let sub = RNodeSubConfig {
887            name: name.to_string(),
888            vport: 0,
889            outgoing: true,
890            frequency,
891            bandwidth,
892            txpower,
893            spreading_factor,
894            coding_rate,
895            flow_control,
896            st_alock,
897            lt_alock,
898        };
899
900        Ok(Box::new(RNodeConfig {
901            name: name.to_string(),
902            port,
903            speed,
904            subinterfaces: vec![sub],
905            multi: false,
906            id_interval,
907            id_callsign,
908            base_interface_id: id,
909            pre_opened_fd,
910            underlay_mark: None,
911            runtime: Arc::new(Mutex::new(RNodeRuntime {
912                sub: RNodeSubConfig {
913                    name: name.to_string(),
914                    vport: 0,
915                    outgoing: true,
916                    frequency,
917                    bandwidth,
918                    txpower,
919                    spreading_factor,
920                    coding_rate,
921                    flow_control,
922                    st_alock,
923                    lt_alock,
924                },
925                writer: None,
926            })),
927        }))
928    }
929
930    fn start(
931        &self,
932        config: Box<dyn InterfaceConfigData>,
933        ctx: StartContext,
934    ) -> std::io::Result<StartResult> {
935        let mut rnode_config = *config.into_any().downcast::<RNodeConfig>().map_err(|_| {
936            std::io::Error::new(std::io::ErrorKind::InvalidData, "wrong config type")
937        })?;
938        rnode_config.underlay_mark = ctx.underlay_mark;
939
940        let sub_bitrates: Vec<u64> = rnode_config
941            .subinterfaces
942            .iter()
943            .map(estimate_lora_bitrate_bps)
944            .collect();
945        let airtime_profiles: Vec<AirtimeProfile> = rnode_config
946            .subinterfaces
947            .iter()
948            .map(lora_airtime_profile)
949            .collect();
950
951        let pairs = start(rnode_config.clone(), ctx.tx)?;
952
953        let mut subs = Vec::with_capacity(pairs.len());
954        for (index, (sub_id, writer)) in pairs.into_iter().enumerate() {
955            let sub_name = rnode_config.subinterfaces[index].name.clone();
956
957            let info = InterfaceInfo {
958                id: sub_id,
959                name: sub_name,
960                mode: ctx.mode,
961                gravity: ctx.gravity,
962                recursive_prs: ctx.recursive_prs,
963                announces_from_internal: ctx.announces_from_internal,
964                announces_to_internal: ctx.announces_to_internal,
965                out_capable: rnode_config.subinterfaces[index].outgoing,
966                in_capable: true,
967                bitrate: sub_bitrates.get(index).copied(),
968                airtime_profile: airtime_profiles.get(index).copied(),
969                announce_rate_target: None,
970                announce_rate_grace: 0,
971                announce_rate_penalty: 0.0,
972                announce_cap: rns_core::constants::ANNOUNCE_CAP,
973                is_local_client: false,
974                wants_tunnel: false,
975                tunnel_id: None,
976                mtu: rns_core::constants::MTU as u32,
977                ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
978                ia_freq: 0.0,
979                ip_freq: 0.0,
980                op_freq: 0.0,
981                op_samples: 0,
982                started: crate::time::now(),
983            };
984
985            subs.push(SubInterface {
986                id: sub_id,
987                info,
988                writer,
989                interface_type_name: if rnode_config.multi {
990                    "RNodeMultiInterface"
991                } else {
992                    "RNodeInterface"
993                }
994                .to_string(),
995            });
996        }
997
998        Ok(StartResult::Multi(subs))
999    }
1000}
1001
1002/// Factory for ConfigObj-driven `RNodeMultiInterface` devices.
1003pub struct RNodeMultiFactory;
1004
1005impl InterfaceFactory for RNodeMultiFactory {
1006    fn type_name(&self) -> &str {
1007        "RNodeMultiInterface"
1008    }
1009
1010    fn default_ifac_size(&self) -> usize {
1011        8
1012    }
1013
1014    fn parse_config(
1015        &self,
1016        name: &str,
1017        id: InterfaceId,
1018        params: &HashMap<String, String>,
1019    ) -> Result<Box<dyn InterfaceConfigData>, String> {
1020        self.parse_config_section(
1021            name,
1022            id,
1023            ConfigSection {
1024                params,
1025                children: &[],
1026            },
1027        )
1028    }
1029
1030    fn parse_config_section(
1031        &self,
1032        name: &str,
1033        id: InterfaceId,
1034        section: ConfigSection<'_>,
1035    ) -> Result<Box<dyn InterfaceConfigData>, String> {
1036        let base = RNodeFactory.parse_config(name, id, section.params)?;
1037        let mut config = *base
1038            .into_any()
1039            .downcast::<RNodeConfig>()
1040            .map_err(|_| "wrong RNode config type".to_string())?;
1041
1042        let mut subinterfaces = Vec::new();
1043        for child in section.children {
1044            let enabled = child
1045                .params
1046                .get("interface_enabled")
1047                .or_else(|| child.params.get("enabled"))
1048                .and_then(|value| crate::config::parse_bool_pub(value))
1049                .unwrap_or(true);
1050            if !enabled {
1051                continue;
1052            }
1053            if subinterfaces.len() >= 11 {
1054                return Err("RNodeMultiInterface supports at most 11 enabled subinterfaces".into());
1055            }
1056            let vport = child
1057                .params
1058                .get("vport")
1059                .ok_or_else(|| format!("subinterface '{}' requires 'vport'", child.name))?
1060                .parse::<u8>()
1061                .map_err(|_| format!("invalid RNode virtual port for '{}'", child.name))?;
1062            if vport > 10
1063                || subinterfaces
1064                    .iter()
1065                    .any(|sub: &RNodeSubConfig| sub.vport == vport)
1066            {
1067                return Err(format!("invalid or duplicate RNode virtual port {}", vport));
1068            }
1069            let parse_u32 = |key: &str, default| {
1070                child
1071                    .params
1072                    .get(key)
1073                    .and_then(|v| v.parse().ok())
1074                    .unwrap_or(default)
1075            };
1076            let parse_u8_alias = |first: &str, second: &str, default| {
1077                child
1078                    .params
1079                    .get(first)
1080                    .or_else(|| child.params.get(second))
1081                    .and_then(|v| v.parse().ok())
1082                    .unwrap_or(default)
1083            };
1084            let sub = RNodeSubConfig {
1085                name: child.name.clone(),
1086                vport,
1087                outgoing: child
1088                    .params
1089                    .get("outgoing")
1090                    .and_then(|v| crate::config::parse_bool_pub(v))
1091                    .unwrap_or(true),
1092                frequency: parse_u32("frequency", 868_000_000),
1093                bandwidth: parse_u32("bandwidth", 125_000),
1094                txpower: child
1095                    .params
1096                    .get("txpower")
1097                    .and_then(|v| v.parse().ok())
1098                    .unwrap_or(7),
1099                spreading_factor: parse_u8_alias("spreadingfactor", "spreading_factor", 8),
1100                coding_rate: parse_u8_alias("codingrate", "coding_rate", 5),
1101                flow_control: child
1102                    .params
1103                    .get("flow_control")
1104                    .and_then(|v| crate::config::parse_bool_pub(v))
1105                    .unwrap_or(false),
1106                st_alock: child
1107                    .params
1108                    .get("airtime_limit_short")
1109                    .or_else(|| child.params.get("st_alock"))
1110                    .and_then(|v| v.parse().ok()),
1111                lt_alock: child
1112                    .params
1113                    .get("airtime_limit_long")
1114                    .or_else(|| child.params.get("lt_alock"))
1115                    .and_then(|v| v.parse().ok()),
1116            };
1117            if let Some(error) = validate_sub_config(&sub) {
1118                return Err(error);
1119            }
1120            subinterfaces.push(sub);
1121        }
1122        if subinterfaces.is_empty() {
1123            return Err("RNodeMultiInterface requires at least one enabled subinterface".into());
1124        }
1125        config.multi = true;
1126        config.subinterfaces = subinterfaces;
1127        *lock_or_recover(&config.runtime, "rnode runtime") = RNodeRuntime::from_config(&config);
1128        Ok(Box::new(config))
1129    }
1130
1131    fn start(
1132        &self,
1133        config: Box<dyn InterfaceConfigData>,
1134        ctx: StartContext,
1135    ) -> io::Result<StartResult> {
1136        RNodeFactory.start(config, ctx)
1137    }
1138}
1139
1140pub(crate) fn rnode_runtime_handle_from_config(config: &RNodeConfig) -> RNodeRuntimeConfigHandle {
1141    RNodeRuntimeConfigHandle {
1142        interface_name: config.name.clone(),
1143        runtime: Arc::clone(&config.runtime),
1144        startup: RNodeRuntime::from_config(config),
1145    }
1146}
1147
1148#[cfg(test)]
1149mod tests {
1150    use super::*;
1151    use crate::event;
1152    use crate::kiss;
1153    use crate::serial::open_pty_pair;
1154    use std::os::unix::io::{AsRawFd, FromRawFd};
1155    use std::path::PathBuf;
1156    use std::sync::mpsc::RecvTimeoutError;
1157    use tempfile::tempdir;
1158    /// Helper: poll an fd for reading with timeout (ms).
1159    fn poll_read(fd: i32, timeout_ms: i32) -> bool {
1160        let mut pfd = libc::pollfd {
1161            fd,
1162            events: libc::POLLIN,
1163            revents: 0,
1164        };
1165        let ret = unsafe { libc::poll(&mut pfd, 1, timeout_ms) };
1166        ret > 0
1167    }
1168
1169    /// Read all available bytes from an fd.
1170    fn read_available(file: &mut Transport) -> Vec<u8> {
1171        let mut all = Vec::new();
1172        let mut buf = [0u8; 4096];
1173        while poll_read(file.as_raw_fd(), 100) {
1174            match file.read(&mut buf) {
1175                Ok(n) if n > 0 => all.extend_from_slice(&buf[..n]),
1176                _ => break,
1177            }
1178        }
1179        all
1180    }
1181
1182    fn slave_tty_path(fd: i32) -> PathBuf {
1183        let mut buf = [0u8; 256];
1184        let rc = unsafe { libc::ttyname_r(fd, buf.as_mut_ptr().cast(), buf.len()) };
1185        assert_eq!(rc, 0, "ttyname_r failed for fd {}", fd);
1186        let nul = buf.iter().position(|b| *b == 0).unwrap_or(buf.len());
1187        PathBuf::from(std::str::from_utf8(&buf[..nul]).unwrap())
1188    }
1189
1190    unsafe fn transport_from_raw_fd(fd: i32) -> Transport {
1191        Transport::Serial(std::fs::File::from_raw_fd(fd))
1192    }
1193
1194    fn wait_for_interface_event<F>(
1195        rx: &crate::event::EventReceiver,
1196        timeout: Duration,
1197        predicate: F,
1198    ) -> Event
1199    where
1200        F: Fn(&Event) -> bool,
1201    {
1202        let deadline = std::time::Instant::now() + timeout;
1203        loop {
1204            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
1205            match rx.recv_timeout(remaining) {
1206                Ok(event) if predicate(&event) => return event,
1207                Ok(_) => continue,
1208                Err(RecvTimeoutError::Timeout) => panic!("timed out waiting for interface event"),
1209                Err(RecvTimeoutError::Disconnected) => panic!("event channel disconnected"),
1210            }
1211        }
1212    }
1213
1214    /// Mock RNode: respond to detect with DETECT_RESP, FW version, platform
1215    fn mock_respond_detect(master: &mut Transport) {
1216        // Respond: DETECT_RESP
1217        master
1218            .write_all(&rnode_kiss::rnode_command(
1219                rnode_kiss::CMD_DETECT,
1220                &[rnode_kiss::DETECT_RESP],
1221            ))
1222            .unwrap();
1223        // FW version 1.74
1224        master
1225            .write_all(&rnode_kiss::rnode_command(
1226                rnode_kiss::CMD_FW_VERSION,
1227                &[0x01, 0x4A],
1228            ))
1229            .unwrap();
1230        // Platform ESP32
1231        master
1232            .write_all(&rnode_kiss::rnode_command(
1233                rnode_kiss::CMD_PLATFORM,
1234                &[0x80],
1235            ))
1236            .unwrap();
1237        // MCU
1238        master
1239            .write_all(&rnode_kiss::rnode_command(rnode_kiss::CMD_MCU, &[0x01]))
1240            .unwrap();
1241        master.flush().unwrap();
1242    }
1243
1244    #[test]
1245    fn rnode_detect_over_pty() {
1246        // Test that the RNode decoder can parse detect responses from a PTY
1247        let (master_fd, slave_fd) = open_pty_pair().unwrap();
1248        let mut master = unsafe { transport_from_raw_fd(master_fd) };
1249        let mut slave = unsafe { transport_from_raw_fd(slave_fd) };
1250
1251        // Write detect response to master
1252        mock_respond_detect(&mut master);
1253
1254        let mut decoder = rnode_kiss::RNodeDecoder::new();
1255        let mut events = Vec::new();
1256        let deadline = std::time::Instant::now() + Duration::from_secs(2);
1257
1258        while std::time::Instant::now() < deadline {
1259            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
1260            let timeout_ms = remaining.as_millis().min(i32::MAX as u128) as i32;
1261            if !poll_read(slave.as_raw_fd(), timeout_ms) {
1262                break;
1263            }
1264
1265            let chunk = read_available(&mut slave);
1266            if chunk.is_empty() {
1267                continue;
1268            }
1269
1270            events.extend(decoder.feed(&chunk));
1271
1272            let saw_detect = events
1273                .iter()
1274                .any(|e| matches!(e, rnode_kiss::RNodeEvent::Detected(true)));
1275            let saw_firmware = events
1276                .iter()
1277                .any(|e| matches!(e, rnode_kiss::RNodeEvent::FirmwareVersion { .. }));
1278            if saw_detect && saw_firmware {
1279                break;
1280            }
1281        }
1282
1283        assert!(
1284            events
1285                .iter()
1286                .any(|e| matches!(e, rnode_kiss::RNodeEvent::Detected(true))),
1287            "should detect device"
1288        );
1289        assert!(
1290            events
1291                .iter()
1292                .any(|e| matches!(e, rnode_kiss::RNodeEvent::FirmwareVersion { .. })),
1293            "should get firmware version"
1294        );
1295    }
1296
1297    #[test]
1298    fn rnode_configure_commands() {
1299        let (master_fd, slave_fd) = open_pty_pair().unwrap();
1300        let mut master = unsafe { transport_from_raw_fd(master_fd) };
1301        let writer_file = unsafe { transport_from_raw_fd(libc::dup(slave_fd)) };
1302        let writer = Arc::new(Mutex::new(writer_file));
1303
1304        let sub = RNodeSubConfig {
1305            name: "test".into(),
1306            vport: 0,
1307            outgoing: true,
1308            frequency: 868_000_000,
1309            bandwidth: 125_000,
1310            txpower: 7,
1311            spreading_factor: 8,
1312            coding_rate: 5,
1313            flow_control: false,
1314            st_alock: None,
1315            lt_alock: None,
1316        };
1317
1318        configure_subinterface(&writer, 0, &sub, false).unwrap();
1319
1320        // Read what was sent
1321        let data = read_available(&mut master);
1322
1323        // Should contain frequency command
1324        assert!(
1325            data.windows(2)
1326                .any(|w| w[0] == kiss::FEND && w[1] == rnode_kiss::CMD_FREQUENCY),
1327            "should contain FREQUENCY command"
1328        );
1329        // Should contain bandwidth command
1330        assert!(
1331            data.windows(2)
1332                .any(|w| w[0] == kiss::FEND && w[1] == rnode_kiss::CMD_BANDWIDTH),
1333            "should contain BANDWIDTH command"
1334        );
1335        // Should contain RADIO_STATE ON
1336        assert!(
1337            data.windows(3).any(|w| w[0] == kiss::FEND
1338                && w[1] == rnode_kiss::CMD_RADIO_STATE
1339                && w[2] == rnode_kiss::RADIO_STATE_ON),
1340            "should contain RADIO_STATE ON command"
1341        );
1342
1343        unsafe { libc::close(slave_fd) };
1344    }
1345
1346    #[test]
1347    fn rnode_data_roundtrip() {
1348        let (master_fd, slave_fd) = open_pty_pair().unwrap();
1349        let mut master = unsafe { transport_from_raw_fd(master_fd) };
1350        let slave = unsafe { transport_from_raw_fd(slave_fd) };
1351
1352        // Write a data frame (subinterface 0) to master
1353        let payload = vec![0x01, 0x02, 0x03, 0x04, 0x05];
1354        let frame = rnode_kiss::rnode_data_frame(0, &payload);
1355        master.write_all(&frame).unwrap();
1356        master.flush().unwrap();
1357
1358        // Read from slave with RNode decoder
1359        assert!(poll_read(slave.as_raw_fd(), 2000));
1360        let mut decoder = rnode_kiss::RNodeDecoder::new();
1361        let mut buf = [0u8; 4096];
1362        let mut slave_file = slave;
1363        let n = slave_file.read(&mut buf).unwrap();
1364        let events = decoder.feed(&buf[..n]);
1365
1366        assert_eq!(events.len(), 1);
1367        match &events[0] {
1368            rnode_kiss::RNodeEvent::DataFrame { index, data } => {
1369                assert_eq!(*index, 0);
1370                assert_eq!(data, &payload);
1371            }
1372            other => panic!("expected DataFrame, got {:?}", other),
1373        }
1374    }
1375
1376    #[test]
1377    fn rnode_flow_control() {
1378        let (master_fd, slave_fd) = open_pty_pair().unwrap();
1379        let writer_file = unsafe { transport_from_raw_fd(slave_fd) };
1380        let shared_writer = Arc::new(Mutex::new(writer_file));
1381
1382        let flow_state = Arc::new(Mutex::new(SubFlowState {
1383            ready: true,
1384            queue: std::collections::VecDeque::new(),
1385        }));
1386
1387        let mut sub_writer = RNodeSubWriter {
1388            writer: shared_writer.clone(),
1389            index: 0,
1390            flow_control: true,
1391            flow_state: flow_state.clone(),
1392            multi: false,
1393        };
1394
1395        // First send: should go through (ready=true) and set ready=false
1396        sub_writer.send_frame(b"hello").unwrap();
1397        assert!(!flow_state.lock().unwrap().ready);
1398
1399        // Second send: should be queued
1400        sub_writer.send_frame(b"world").unwrap();
1401        assert_eq!(flow_state.lock().unwrap().queue.len(), 1);
1402
1403        // Process flow queue (simulates CMD_READY)
1404        process_flow_queue(&flow_state, &shared_writer, 0, false);
1405        assert_eq!(flow_state.lock().unwrap().queue.len(), 0);
1406        assert!(!flow_state.lock().unwrap().ready); // sent queued, still locked
1407
1408        // Process again with empty queue: sets ready=true
1409        process_flow_queue(&flow_state, &shared_writer, 0, false);
1410        assert!(flow_state.lock().unwrap().ready);
1411
1412        unsafe { libc::close(master_fd) };
1413    }
1414
1415    #[test]
1416    fn rnode_reset_clears_pending_rx_metadata() {
1417        let mut last_rssi = Some(-101);
1418        let mut last_snr = Some(7.25);
1419
1420        clear_pending_rx_metadata(&mut last_rssi, &mut last_snr);
1421
1422        assert_eq!(last_rssi, None);
1423        assert_eq!(last_snr, None);
1424    }
1425
1426    #[test]
1427    fn rnode_snr_metadata_does_not_depend_on_rssi() {
1428        let mut last_rssi = None;
1429        let mut last_snr = None;
1430
1431        assert!(update_pending_rx_metadata(
1432            &rnode_kiss::RNodeEvent::StatSnr(-12),
1433            &mut last_rssi,
1434            &mut last_snr,
1435        ));
1436
1437        assert_eq!(last_rssi, None);
1438        assert_eq!(last_snr, Some(-3.0));
1439    }
1440
1441    #[test]
1442    fn rnode_sub_writer_format() {
1443        let (master_fd, slave_fd) = open_pty_pair().unwrap();
1444        let mut master = unsafe { transport_from_raw_fd(master_fd) };
1445        let writer_file = unsafe { transport_from_raw_fd(slave_fd) };
1446        let shared_writer = Arc::new(Mutex::new(writer_file));
1447
1448        let flow_state = Arc::new(Mutex::new(SubFlowState {
1449            ready: true,
1450            queue: std::collections::VecDeque::new(),
1451        }));
1452
1453        let mut sub_writer = RNodeSubWriter {
1454            writer: shared_writer,
1455            index: 1, // subinterface 1
1456            flow_control: false,
1457            flow_state,
1458            multi: false,
1459        };
1460
1461        let payload = vec![0xAA, 0xBB, 0xCC];
1462        sub_writer.send_frame(&payload).unwrap();
1463
1464        // Read from master
1465        assert!(poll_read(master.as_raw_fd(), 2000));
1466        let mut buf = [0u8; 256];
1467        let n = master.read(&mut buf).unwrap();
1468
1469        // Should start with FEND, CMD_INT1_DATA (0x10)
1470        assert_eq!(buf[0], kiss::FEND);
1471        assert_eq!(buf[1], 0x10); // CMD_INT1_DATA
1472        assert_eq!(buf[n - 1], kiss::FEND);
1473    }
1474
1475    #[test]
1476    fn rnode_multi_sub_routing() {
1477        // Test that data frames on different CMD_INTn route to different indices
1478        let mut decoder = rnode_kiss::RNodeDecoder::new();
1479
1480        let payload0 = vec![0x01, 0x02];
1481        let frame0 = rnode_kiss::rnode_data_frame(0, &payload0);
1482        let events0 = decoder.feed(&frame0);
1483        assert_eq!(events0.len(), 1);
1484        assert_eq!(
1485            events0[0],
1486            rnode_kiss::RNodeEvent::DataFrame {
1487                index: 0,
1488                data: payload0
1489            }
1490        );
1491
1492        let payload1 = vec![0x03, 0x04];
1493        let frame1 = rnode_kiss::rnode_data_frame(1, &payload1);
1494        let events1 = decoder.feed(&frame1);
1495        assert_eq!(events1.len(), 1);
1496        assert_eq!(
1497            events1[0],
1498            rnode_kiss::RNodeEvent::DataFrame {
1499                index: 1,
1500                data: payload1
1501            }
1502        );
1503    }
1504
1505    #[test]
1506    fn rnode_error_handling() {
1507        // NOTE: CMD_ERROR (0x90) == CMD_INT5_DATA (0x90) — they share the same
1508        // byte value. In multi-radio mode, 0x90 is treated as INT5_DATA.
1509        // Error events only appear on single-radio devices where there's no
1510        // INT5_DATA. The decoder treats 0x90 as INT5_DATA (data wins).
1511        // For real error detection, we rely on the data frame being invalid
1512        // or the device sending a different error indicator.
1513        // Test that cmd 0x90 is correctly handled as a data frame.
1514        let mut decoder = rnode_kiss::RNodeDecoder::new();
1515        let frame = rnode_kiss::rnode_command(rnode_kiss::CMD_ERROR, &[0x02]);
1516        let events = decoder.feed(&frame);
1517        assert_eq!(events.len(), 1);
1518        // 0x90 = CMD_INT5_DATA, so it's a DataFrame with index 5
1519        assert_eq!(
1520            events[0],
1521            rnode_kiss::RNodeEvent::DataFrame {
1522                index: 5,
1523                data: vec![0x02]
1524            }
1525        );
1526    }
1527
1528    #[test]
1529    fn rnode_config_validation() {
1530        let good = RNodeSubConfig {
1531            name: "test".into(),
1532            vport: 0,
1533            outgoing: true,
1534            frequency: 868_000_000,
1535            bandwidth: 125_000,
1536            txpower: 7,
1537            spreading_factor: 8,
1538            coding_rate: 5,
1539            flow_control: false,
1540            st_alock: None,
1541            lt_alock: None,
1542        };
1543        assert!(validate_sub_config(&good).is_none());
1544
1545        // Bad frequency
1546        let mut bad = good.clone();
1547        bad.frequency = 100;
1548        assert!(validate_sub_config(&bad).is_some());
1549
1550        // Bad SF
1551        bad = good.clone();
1552        bad.spreading_factor = 13;
1553        assert!(validate_sub_config(&bad).is_some());
1554
1555        // Bad CR
1556        bad = good.clone();
1557        bad.coding_rate = 9;
1558        assert!(validate_sub_config(&bad).is_some());
1559
1560        // Bad BW
1561        bad = good.clone();
1562        bad.bandwidth = 5;
1563        assert!(validate_sub_config(&bad).is_some());
1564
1565        // Bad TX power
1566        bad = good.clone();
1567        bad.txpower = 50;
1568        assert!(validate_sub_config(&bad).is_some());
1569    }
1570
1571    #[test]
1572    fn multi_factory_preserves_nonsequential_virtual_ports_and_disabled_children() {
1573        let parent = HashMap::from([
1574            ("port".to_string(), "/dev/ttyUSB0".to_string()),
1575            ("speed".to_string(), "115200".to_string()),
1576        ]);
1577        let children = vec![
1578            crate::config::ParsedSubinterface {
1579                name: "disabled".into(),
1580                params: HashMap::from([
1581                    ("interface_enabled".into(), "no".into()),
1582                    ("vport".into(), "2".into()),
1583                ]),
1584            },
1585            crate::config::ParsedSubinterface {
1586                name: "high".into(),
1587                params: HashMap::from([
1588                    ("interface_enabled".into(), "yes".into()),
1589                    ("vport".into(), "7".into()),
1590                    ("frequency".into(), "915000000".into()),
1591                    ("airtime_limit_short".into(), "12.5".into()),
1592                    ("outgoing".into(), "no".into()),
1593                ]),
1594            },
1595            crate::config::ParsedSubinterface {
1596                name: "low".into(),
1597                params: HashMap::from([
1598                    ("enabled".into(), "yes".into()),
1599                    ("vport".into(), "1".into()),
1600                ]),
1601            },
1602        ];
1603        let parsed = RNodeMultiFactory
1604            .parse_config_section(
1605                "multi",
1606                InterfaceId(40),
1607                ConfigSection {
1608                    params: &parent,
1609                    children: &children,
1610                },
1611            )
1612            .unwrap();
1613        let config = parsed.into_any().downcast::<RNodeConfig>().unwrap();
1614        assert_eq!(config.subinterfaces.len(), 2);
1615        assert_eq!(config.subinterfaces[0].vport, 7);
1616        assert_eq!(config.subinterfaces[1].vport, 1);
1617        assert!(!config.subinterfaces[0].outgoing);
1618        assert_eq!(config.subinterfaces[0].st_alock, Some(12.5));
1619    }
1620
1621    #[test]
1622    fn rnode_lora_bitrate_estimate_uses_radio_params() {
1623        let mut sub = RNodeSubConfig {
1624            name: "test".into(),
1625            vport: 0,
1626            outgoing: true,
1627            frequency: 868_000_000,
1628            bandwidth: 125_000,
1629            txpower: 7,
1630            spreading_factor: 8,
1631            coding_rate: 5,
1632            flow_control: false,
1633            st_alock: None,
1634            lt_alock: None,
1635        };
1636
1637        assert_eq!(estimate_lora_bitrate_bps(&sub), 3125);
1638
1639        sub.spreading_factor = 12;
1640        assert_eq!(estimate_lora_bitrate_bps(&sub), 293);
1641
1642        sub.bandwidth = 250_000;
1643        assert_eq!(estimate_lora_bitrate_bps(&sub), 586);
1644    }
1645
1646    #[test]
1647    fn rnode_lora_airtime_profile_uses_radio_params() {
1648        let sub = RNodeSubConfig {
1649            name: "test".into(),
1650            vport: 0,
1651            outgoing: true,
1652            frequency: 868_000_000,
1653            bandwidth: 125_000,
1654            txpower: 7,
1655            spreading_factor: 8,
1656            coding_rate: 5,
1657            flow_control: false,
1658            st_alock: None,
1659            lt_alock: None,
1660        };
1661
1662        let profile = lora_airtime_profile(&sub);
1663        assert!((profile.transmit_time_secs(100) - 0.307712).abs() < 0.000001);
1664        assert_eq!(
1665            profile,
1666            AirtimeProfile::Lora {
1667                bandwidth: 125_000,
1668                spreading_factor: 8,
1669                coding_rate: 5,
1670                preamble_symbols: LORA_PREAMBLE_SYMBOLS,
1671                explicit_header: LORA_EXPLICIT_HEADER,
1672                crc: LORA_CRC,
1673            }
1674        );
1675    }
1676
1677    #[test]
1678    fn rnode_reconnects_after_serial_disconnect() {
1679        let tempdir = tempdir().unwrap();
1680        let port_path = tempdir.path().join("rnode-port");
1681
1682        let (master1_fd, slave1_fd) = open_pty_pair().unwrap();
1683        let slave1_path = slave_tty_path(slave1_fd);
1684        std::os::unix::fs::symlink(&slave1_path, &port_path).unwrap();
1685
1686        let mut master1 = unsafe { transport_from_raw_fd(master1_fd) };
1687        let slave1 = unsafe { transport_from_raw_fd(slave1_fd) };
1688
1689        let (tx, rx) = event::channel();
1690        let sub = RNodeSubConfig {
1691            name: "test-rnode".into(),
1692            vport: 0,
1693            outgoing: true,
1694            frequency: 868_000_000,
1695            bandwidth: 125_000,
1696            txpower: 7,
1697            spreading_factor: 8,
1698            coding_rate: 5,
1699            flow_control: false,
1700            st_alock: None,
1701            lt_alock: None,
1702        };
1703        let mut config = RNodeConfig {
1704            name: "test-rnode".into(),
1705            port: port_path.display().to_string(),
1706            speed: 115200,
1707            subinterfaces: vec![sub],
1708            multi: false,
1709            id_interval: None,
1710            id_callsign: None,
1711            base_interface_id: InterfaceId(41),
1712            pre_opened_fd: None,
1713            underlay_mark: None,
1714            runtime: Arc::new(Mutex::new(RNodeRuntime {
1715                sub: RNodeSubConfig {
1716                    name: String::new(),
1717                    vport: 0,
1718                    outgoing: true,
1719                    frequency: 868_000_000,
1720                    bandwidth: 125_000,
1721                    txpower: 7,
1722                    spreading_factor: 8,
1723                    coding_rate: 5,
1724                    flow_control: false,
1725                    st_alock: None,
1726                    lt_alock: None,
1727                },
1728                writer: None,
1729            })),
1730        };
1731        config.runtime = Arc::new(Mutex::new(RNodeRuntime::from_config(&config)));
1732
1733        let _writers = start(config, tx).unwrap();
1734
1735        thread::sleep(Duration::from_secs(3));
1736        mock_respond_detect(&mut master1);
1737        let up = wait_for_interface_event(&rx, Duration::from_secs(4), |event| {
1738            matches!(event, Event::InterfaceUp(InterfaceId(41), _, _))
1739        });
1740        assert!(matches!(
1741            up,
1742            Event::InterfaceUp(InterfaceId(41), None, None)
1743        ));
1744
1745        drop(master1);
1746        drop(slave1);
1747
1748        let down = wait_for_interface_event(&rx, Duration::from_secs(4), |event| {
1749            matches!(event, Event::InterfaceDown(InterfaceId(41)))
1750        });
1751        assert!(matches!(down, Event::InterfaceDown(InterfaceId(41))));
1752
1753        let (master2_fd, slave2_fd) = open_pty_pair().unwrap();
1754        let slave2_path = slave_tty_path(slave2_fd);
1755        std::fs::remove_file(&port_path).unwrap();
1756        std::os::unix::fs::symlink(&slave2_path, &port_path).unwrap();
1757
1758        let mut master2 = unsafe { transport_from_raw_fd(master2_fd) };
1759        let _slave2 = unsafe { transport_from_raw_fd(slave2_fd) };
1760
1761        thread::sleep(Duration::from_secs(3));
1762        mock_respond_detect(&mut master2);
1763        let up = wait_for_interface_event(&rx, Duration::from_secs(4), |event| {
1764            matches!(event, Event::InterfaceUp(InterfaceId(41), _, _))
1765        });
1766        assert!(matches!(
1767            up,
1768            Event::InterfaceUp(InterfaceId(41), Some(_), None)
1769        ));
1770    }
1771}