prns_interfaces_embassy/
esp_now.rs

1use alloc::boxed::Box;
2
3use embassy_futures::select::{select3, Either3};
4use embassy_time::Instant;
5use heapless::Vec as HeaplessVec;
6
7use prns_core::engine::InstantMillis;
8use prns_core::interfaces::esp_now::{
9    self, ChannelPolicy, EspNowRadio, CHANNEL_TAG_CAP, ESP_NOW_HW_MTU, ESP_NOW_V2_AIR_MTU,
10};
11use prns_core::interfaces::{
12    BitrateBps, ConnectionState, InterfaceDescriptor, InterfaceId, InterfaceKind,
13};
14use prns_runtime::manifold::driver::EmbassyInterfaceStatus;
15use prns_runtime::manifold::interface_seam::{
16    Interface, InterfaceSeam, OutboundDisposition, OutboundDropReason,
17};
18use prns_runtime::manifold::throughput::ThroughputLedger;
19
20pub struct EspNowInterface<'a, R> {
21    id: InterfaceId,
22    radio: R,
23    policy: ChannelPolicy,
24    bitrate: BitrateBps,
25    tag: HeaplessVec<u8, CHANNEL_TAG_CAP>,
26    status: &'a EmbassyInterfaceStatus,
27}
28
29impl<'a, R> EspNowInterface<'a, R> {
30    #[must_use]
31    pub fn new(
32        radio: R,
33        policy: ChannelPolicy,
34        bitrate: BitrateBps,
35        status: &'a EmbassyInterfaceStatus,
36    ) -> Self {
37        Self {
38            id: esp_now::interface_id(),
39            radio,
40            policy,
41            bitrate,
42            tag: esp_now::channel_tag(),
43            status,
44        }
45    }
46
47    #[must_use]
48    pub fn id(&self) -> InterfaceId {
49        self.id
50    }
51
52    /// The id this interface will carry — for the caller that stands its [`EmbassyInterfaceStatus`] up under the same key before building the interface.
53    #[must_use]
54    pub fn interface_id() -> InterfaceId {
55        esp_now::interface_id()
56    }
57}
58
59impl<R: EspNowRadio> Interface for EspNowInterface<'_, R> {
60    const HW_MTU: usize = ESP_NOW_HW_MTU;
61    const KIND: InterfaceKind = InterfaceKind::EspNow;
62
63    fn descriptor(&self) -> InterfaceDescriptor {
64        esp_now::descriptor(self.id, self.bitrate)
65    }
66
67    fn channel_tag(&self) -> &[u8] {
68        &self.tag
69    }
70
71    async fn run<Seam: InterfaceSeam>(self, mut seam: Seam) {
72        let EspNowInterface {
73            mut radio,
74            policy,
75            status,
76            ..
77        } = self;
78        if let ChannelPolicy::Fixed(channel) = policy {
79            radio.set_channel(channel);
80        }
81
82        let mut rx_buf = Box::new([0u8; ESP_NOW_V2_AIR_MTU]);
83        let mut throughput = ThroughputLedger::new();
84        let started = Instant::now();
85        status.set_connection(ConnectionState::Connected);
86        crate::diagnostic_log::info!("RNS_ESPNOW interface up, policy {policy:?}");
87
88        loop {
89            if !status.is_enabled() {
90                status.set_connection(ConnectionState::Disabled);
91                status.wait_until_enabled().await;
92                status.set_connection(ConnectionState::Connected);
93            }
94
95            match select3(
96                radio.receive(&mut rx_buf[..]),
97                seam.next_outbound(),
98                status.wait_until_disabled(),
99            )
100            .await
101            {
102                Either3::First(len) => {
103                    if len > 0 {
104                        let now = InstantMillis(started.elapsed().as_millis());
105                        status.add_rx(len as u64);
106                        throughput.record_rx(now, len as u64);
107                        status.set_transfer_rates(throughput.rates());
108                        seam.next_inbound(&rx_buf[..len]).await;
109                    }
110                }
111                Either3::Second(outbound) => {
112                    let len = outbound.len().min(ESP_NOW_V2_AIR_MTU);
113                    let disposition = if radio.broadcast(&outbound[..len]).await {
114                        let now = InstantMillis(started.elapsed().as_millis());
115                        status.add_tx(len as u64);
116                        throughput.record_tx(now, len as u64);
117                        status.set_transfer_rates(throughput.rates());
118                        OutboundDisposition::Sent
119                    } else {
120                        OutboundDisposition::Dropped(OutboundDropReason::TransportFailure)
121                    };
122                    seam.complete_outbound(disposition);
123                }
124                Either3::Third(()) => {}
125            }
126        }
127    }
128}
prns_interfaces_embassy/esp_now.rs

prns_interfaces_embassy/
esp_now.rs
