pamoja_gateway/udp/
payload.rs

1//! The JSON objects the datagrams carry: `rxpk`, `stat`, `txpk`, and `txpk_ack`.
2//!
3//! The protocol writes a frequency as megahertz with hertz precision, a payload as base64, a
4//! LoRa link as a datarate identifier such as `SF7BW125` beside a coding rate such as `4/5`,
5//! and every level as a plain number. These types hold each of those the way the rest of the
6//! SDK does, in hertz, in bytes, as [`LinkSettings`], and in [`Decibels`], and convert at the
7//! boundary.
8
9use std::fmt;
10
11use pamoja_lora::budget::Decibels;
12use pamoja_lora::LinkSettings;
13use serde_json::{json, Map, Value};
14
15use crate::base64;
16
17use super::ProtocolError;
18
19/// What the CRC of a received packet said.
20#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
21pub enum CrcStatus {
22    /// The CRC checked, which is `1`.
23    #[default]
24    Ok,
25    /// The CRC failed, which is `-1`.
26    Failed,
27    /// The packet carried no CRC, which is `0`.
28    Absent,
29}
30
31impl CrcStatus {
32    /// Returns the number the protocol writes.
33    ///
34    /// # Returns
35    ///
36    /// `1`, `-1`, or `0`.
37    pub const fn code(self) -> i8 {
38        match self {
39            CrcStatus::Ok => 1,
40            CrcStatus::Failed => -1,
41            CrcStatus::Absent => 0,
42        }
43    }
44
45    /// Names the status a number selects.
46    ///
47    /// # Arguments
48    ///
49    /// * `code` - the number the protocol wrote.
50    ///
51    /// # Returns
52    ///
53    /// The status, or `None` for any other number.
54    pub const fn from_code(code: i64) -> Option<CrcStatus> {
55        match code {
56            1 => Some(CrcStatus::Ok),
57            -1 => Some(CrcStatus::Failed),
58            0 => Some(CrcStatus::Absent),
59            _ => None,
60        }
61    }
62}
63
64/// How a packet was modulated.
65#[derive(Clone, Copy, Debug, PartialEq, Eq)]
66pub enum Modulation {
67    /// LoRa, whose datarate identifier and coding rate are the link's own settings.
68    Lora(LinkSettings),
69    /// FSK, at a bitrate in bits per second.
70    Fsk(u32),
71}
72
73impl Modulation {
74    /// Returns the link settings, for a LoRa packet.
75    ///
76    /// # Returns
77    ///
78    /// The settings, or `None` for FSK.
79    pub const fn link(self) -> Option<LinkSettings> {
80        match self {
81            Modulation::Lora(link) => Some(link),
82            Modulation::Fsk(_) => None,
83        }
84    }
85
86    /// Writes the datarate identifier, such as `SF7BW125`.
87    ///
88    /// # Returns
89    ///
90    /// The identifier for LoRa, or the bitrate as a number for FSK.
91    fn datarate(self) -> Value {
92        match self {
93            Modulation::Lora(link) => Value::String(format!(
94                "SF{}BW{}",
95                link.spreading_factor(),
96                link.bandwidth_hz() / 1_000
97            )),
98            Modulation::Fsk(bitrate) => json!(bitrate),
99        }
100    }
101}
102
103/// Reads a datarate identifier such as `SF7BW125`.
104///
105/// # Arguments
106///
107/// * `text` - the identifier.
108///
109/// # Returns
110///
111/// The spreading factor and bandwidth it names, or `None` when it is not that shape.
112fn link_of(text: &str) -> Option<(u8, u32)> {
113    let (factor, bandwidth) = text.strip_prefix("SF")?.split_once("BW")?;
114    Some((
115        factor.parse().ok()?,
116        bandwidth.parse::<u32>().ok()?.checked_mul(1_000)?,
117    ))
118}
119
120/// Reads a coding rate such as `4/5`.
121///
122/// # Arguments
123///
124/// * `text` - the coding rate.
125///
126/// # Returns
127///
128/// Its denominator, or `None` when the numerator is not 4 or the text is not that shape.
129fn coding_rate_of(text: &str) -> Option<u8> {
130    let (numerator, denominator) = text.split_once('/')?;
131    if numerator != "4" {
132        return None;
133    }
134    denominator.parse().ok()
135}
136
137/// A packet the gateway heard, with the metadata the protocol carries beside it.
138///
139/// # Examples
140///
141/// ```
142/// use pamoja_gateway::udp::Rxpk;
143/// use pamoja_lora::LinkSettings;
144///
145/// let heard = Rxpk::new(868_100_000, LinkSettings::new(7, 125_000), b"hello".to_vec())
146///     .with_rssi_dbm(-35)
147///     .with_snr_db(5.1);
148/// assert!(heard.to_json().contains("\"datr\":\"SF7BW125\""));
149/// assert!(heard.to_json().contains("\"freq\":868.1"));
150/// ```
151#[derive(Clone, Debug, PartialEq)]
152pub struct Rxpk {
153    /// When the packet arrived, as [`time::compact`](crate::time::compact) writes it.
154    pub received_at: Option<String>,
155    /// When it arrived on the GPS clock, in milliseconds since 6 January 1980.
156    pub gps_millis: Option<u64>,
157    /// The concentrator's own timestamp of the end of reception, in microseconds.
158    pub timestamp_us: Option<u32>,
159    /// The carrier the packet arrived on, in hertz.
160    pub frequency_hz: u32,
161    /// The concentrator channel it arrived on.
162    pub channel: u8,
163    /// The radio chain it arrived on.
164    pub rf_chain: u8,
165    /// What the CRC said.
166    pub crc: CrcStatus,
167    /// How it was modulated.
168    pub modulation: Modulation,
169    /// The received signal strength, to the decibel.
170    pub rssi_dbm: Decibels,
171    /// The signal-to-noise ratio, to a tenth of a decibel, for a LoRa packet.
172    pub snr_db: Option<Decibels>,
173    /// The packet itself.
174    pub payload: Vec<u8>,
175}
176
177impl Rxpk {
178    /// Describes a LoRa packet that arrived with a good CRC.
179    ///
180    /// # Arguments
181    ///
182    /// * `frequency_hz` - the carrier it arrived on.
183    /// * `link` - the spreading factor, bandwidth, and coding rate it used.
184    /// * `payload` - the packet.
185    ///
186    /// # Returns
187    ///
188    /// The description, on channel 0 and radio chain 0 with no levels yet.
189    pub fn new(frequency_hz: u32, link: LinkSettings, payload: Vec<u8>) -> Rxpk {
190        Rxpk {
191            received_at: None,
192            gps_millis: None,
193            timestamp_us: None,
194            frequency_hz,
195            channel: 0,
196            rf_chain: 0,
197            crc: CrcStatus::Ok,
198            modulation: Modulation::Lora(link),
199            rssi_dbm: Decibels::ZERO,
200            snr_db: None,
201            payload,
202        }
203    }
204
205    /// Returns it with the signal strength it was heard at.
206    ///
207    /// # Arguments
208    ///
209    /// * `dbm` - the strength in dBm.
210    ///
211    /// # Returns
212    ///
213    /// The description.
214    pub fn with_rssi_dbm(mut self, dbm: i32) -> Rxpk {
215        self.rssi_dbm = Decibels::from_db(dbm);
216        self
217    }
218
219    /// Returns it with the signal-to-noise ratio it was heard at.
220    ///
221    /// # Arguments
222    ///
223    /// * `db` - the ratio in decibels, which the protocol carries to a tenth.
224    ///
225    /// # Returns
226    ///
227    /// The description.
228    pub fn with_snr_db(mut self, db: f64) -> Rxpk {
229        self.snr_db = Some(Decibels::from_hundredths((db * 100.0).round() as i32));
230        self
231    }
232
233    /// Returns it with the concentrator's timestamp of the reception.
234    ///
235    /// # Arguments
236    ///
237    /// * `micros` - the counter value, which wraps every 71 minutes.
238    ///
239    /// # Returns
240    ///
241    /// The description.
242    pub fn with_timestamp_us(mut self, micros: u32) -> Rxpk {
243        self.timestamp_us = Some(micros);
244        self
245    }
246
247    /// Returns it with the time it arrived.
248    ///
249    /// # Arguments
250    ///
251    /// * `micros_since_epoch` - the time in microseconds since 1970-01-01 UTC.
252    ///
253    /// # Returns
254    ///
255    /// The description.
256    pub fn with_received_at(mut self, micros_since_epoch: u64) -> Rxpk {
257        self.received_at = Some(crate::time::compact(micros_since_epoch));
258        self
259    }
260
261    /// Returns it on another concentrator channel and radio chain.
262    ///
263    /// # Arguments
264    ///
265    /// * `channel` - the concentrator channel.
266    /// * `rf_chain` - the radio chain.
267    ///
268    /// # Returns
269    ///
270    /// The description.
271    pub fn on_channel(mut self, channel: u8, rf_chain: u8) -> Rxpk {
272        self.channel = channel;
273        self.rf_chain = rf_chain;
274        self
275    }
276
277    /// Writes the object as the protocol carries it.
278    ///
279    /// # Returns
280    ///
281    /// The JSON text of one `rxpk` entry.
282    pub fn to_json(&self) -> String {
283        self.to_value().to_string()
284    }
285
286    /// Builds the object the protocol carries.
287    fn to_value(&self) -> Value {
288        let mut object = Map::new();
289        if let Some(time) = &self.received_at {
290            object.insert("time".to_owned(), json!(time));
291        }
292        if let Some(millis) = self.gps_millis {
293            object.insert("tmms".to_owned(), json!(millis));
294        }
295        if let Some(micros) = self.timestamp_us {
296            object.insert("tmst".to_owned(), json!(micros));
297        }
298        object.insert("chan".to_owned(), json!(self.channel));
299        object.insert("rfch".to_owned(), json!(self.rf_chain));
300        object.insert("freq".to_owned(), megahertz(self.frequency_hz));
301        object.insert("stat".to_owned(), json!(self.crc.code()));
302        object.insert("modu".to_owned(), json!(name_of(self.modulation)));
303        object.insert("datr".to_owned(), self.modulation.datarate());
304        if let Modulation::Lora(link) = self.modulation {
305            object.insert(
306                "codr".to_owned(),
307                json!(format!("4/{}", link.coding_rate_denominator())),
308            );
309        }
310        object.insert("rssi".to_owned(), json!(self.rssi_dbm.round_db()));
311        if let Some(snr) = self.snr_db {
312            object.insert("lsnr".to_owned(), decibels(snr));
313        }
314        object.insert("size".to_owned(), json!(self.payload.len()));
315        object.insert("data".to_owned(), json!(base64::encode(&self.payload)));
316        Value::Object(object)
317    }
318
319    /// Reads one `rxpk` entry.
320    fn from_value(value: &Value) -> Result<Rxpk, ProtocolError> {
321        let object = object_of(value, "an rxpk entry")?;
322        let modulation = modulation_of(object, "rxpk")?;
323        Ok(Rxpk {
324            received_at: text(object, "time").map(str::to_owned),
325            gps_millis: whole(object, "tmms").map(|value| value as u64),
326            timestamp_us: whole(object, "tmst").map(|value| value as u32),
327            frequency_hz: hertz(object, "rxpk")?,
328            channel: whole(object, "chan").unwrap_or(0) as u8,
329            rf_chain: whole(object, "rfch").unwrap_or(0) as u8,
330            crc: whole(object, "stat")
331                .and_then(CrcStatus::from_code)
332                .unwrap_or(CrcStatus::Absent),
333            modulation,
334            rssi_dbm: Decibels::from_db(whole(object, "rssi").unwrap_or(0) as i32),
335            snr_db: number(object, "lsnr")
336                .map(|snr| Decibels::from_hundredths((snr * 100.0).round() as i32)),
337            payload: payload_of(object, "rxpk")?,
338        })
339    }
340}
341
342/// The gateway's own status report.
343///
344/// # Examples
345///
346/// ```
347/// use pamoja_gateway::udp::Stat;
348///
349/// let report = Stat::new().with_counts(2, 2, 2).with_downlinks(2, 2);
350/// assert!(report.to_json().contains("\"rxfw\":2"));
351/// ```
352#[derive(Clone, Debug, Default, PartialEq)]
353pub struct Stat {
354    /// The gateway's clock, as [`time::expanded`](crate::time::expanded) writes it.
355    pub time: Option<String>,
356    /// Its latitude in degrees, north positive.
357    pub latitude_deg: Option<f64>,
358    /// Its longitude in degrees, east positive.
359    pub longitude_deg: Option<f64>,
360    /// Its altitude in meters.
361    pub altitude_m: Option<i32>,
362    /// How many packets its radio received.
363    pub received: u32,
364    /// How many of those had a good CRC.
365    pub received_ok: u32,
366    /// How many it forwarded.
367    pub forwarded: u32,
368    /// What share of its datagrams were acknowledged, as a percentage.
369    pub acknowledged_percent: f64,
370    /// How many downlink datagrams it received.
371    pub downlinks: u32,
372    /// How many packets it transmitted.
373    pub transmitted: u32,
374}
375
376impl Stat {
377    /// An empty report, which is what a gateway with nothing yet to say sends.
378    ///
379    /// # Returns
380    ///
381    /// The report.
382    pub fn new() -> Stat {
383        Stat::default()
384    }
385
386    /// Returns it with the gateway's clock.
387    ///
388    /// # Arguments
389    ///
390    /// * `seconds_since_epoch` - the time in seconds since 1970-01-01 UTC.
391    ///
392    /// # Returns
393    ///
394    /// The report.
395    pub fn at(mut self, seconds_since_epoch: u64) -> Stat {
396        self.time = Some(crate::time::expanded(seconds_since_epoch));
397        self
398    }
399
400    /// Returns it with the gateway's position.
401    ///
402    /// # Arguments
403    ///
404    /// * `latitude_deg` - degrees north.
405    /// * `longitude_deg` - degrees east.
406    /// * `altitude_m` - meters above sea level.
407    ///
408    /// # Returns
409    ///
410    /// The report.
411    pub fn at_position(mut self, latitude_deg: f64, longitude_deg: f64, altitude_m: i32) -> Stat {
412        self.latitude_deg = Some(latitude_deg);
413        self.longitude_deg = Some(longitude_deg);
414        self.altitude_m = Some(altitude_m);
415        self
416    }
417
418    /// Returns it with what the radio heard.
419    ///
420    /// # Arguments
421    ///
422    /// * `received` - packets received.
423    /// * `received_ok` - those with a good CRC.
424    /// * `forwarded` - those forwarded to the server.
425    ///
426    /// # Returns
427    ///
428    /// The report.
429    pub fn with_counts(mut self, received: u32, received_ok: u32, forwarded: u32) -> Stat {
430        self.received = received;
431        self.received_ok = received_ok;
432        self.forwarded = forwarded;
433        self
434    }
435
436    /// Returns it with what the downlink side did.
437    ///
438    /// # Arguments
439    ///
440    /// * `downlinks` - datagrams received from the server.
441    /// * `transmitted` - packets transmitted.
442    ///
443    /// # Returns
444    ///
445    /// The report.
446    pub fn with_downlinks(mut self, downlinks: u32, transmitted: u32) -> Stat {
447        self.downlinks = downlinks;
448        self.transmitted = transmitted;
449        self
450    }
451
452    /// Returns it with the share of its datagrams the server acknowledged.
453    ///
454    /// # Arguments
455    ///
456    /// * `percent` - the percentage.
457    ///
458    /// # Returns
459    ///
460    /// The report.
461    pub fn with_acknowledged_percent(mut self, percent: f64) -> Stat {
462        self.acknowledged_percent = percent;
463        self
464    }
465
466    /// Writes the object as the protocol carries it.
467    ///
468    /// # Returns
469    ///
470    /// The JSON text of the `stat` object.
471    pub fn to_json(&self) -> String {
472        self.to_value().to_string()
473    }
474
475    /// Builds the object the protocol carries.
476    fn to_value(&self) -> Value {
477        let mut object = Map::new();
478        if let Some(time) = &self.time {
479            object.insert("time".to_owned(), json!(time));
480        }
481        if let (Some(latitude), Some(longitude)) = (self.latitude_deg, self.longitude_deg) {
482            object.insert("lati".to_owned(), json!(latitude));
483            object.insert("long".to_owned(), json!(longitude));
484        }
485        if let Some(altitude) = self.altitude_m {
486            object.insert("alti".to_owned(), json!(altitude));
487        }
488        object.insert("rxnb".to_owned(), json!(self.received));
489        object.insert("rxok".to_owned(), json!(self.received_ok));
490        object.insert("rxfw".to_owned(), json!(self.forwarded));
491        object.insert("ackr".to_owned(), json!(self.acknowledged_percent));
492        object.insert("dwnb".to_owned(), json!(self.downlinks));
493        object.insert("txnb".to_owned(), json!(self.transmitted));
494        Value::Object(object)
495    }
496
497    /// Reads the `stat` object.
498    fn from_value(value: &Value) -> Result<Stat, ProtocolError> {
499        let object = object_of(value, "a stat object")?;
500        Ok(Stat {
501            time: text(object, "time").map(str::to_owned),
502            latitude_deg: number(object, "lati"),
503            longitude_deg: number(object, "long"),
504            altitude_m: whole(object, "alti").map(|value| value as i32),
505            received: whole(object, "rxnb").unwrap_or(0) as u32,
506            received_ok: whole(object, "rxok").unwrap_or(0) as u32,
507            forwarded: whole(object, "rxfw").unwrap_or(0) as u32,
508            acknowledged_percent: number(object, "ackr").unwrap_or(0.0),
509            downlinks: whole(object, "dwnb").unwrap_or(0) as u32,
510            transmitted: whole(object, "txnb").unwrap_or(0) as u32,
511        })
512    }
513}
514
515/// What a PUSH_DATA carries: the packets heard, and the gateway's own report.
516#[derive(Clone, Debug, Default, PartialEq)]
517pub struct Uplink {
518    /// The packets, which may be none when only a report is being sent.
519    pub packets: Vec<Rxpk>,
520    /// The report, which a gateway sends every half minute or so.
521    pub status: Option<Stat>,
522}
523
524impl From<Rxpk> for Uplink {
525    fn from(packet: Rxpk) -> Uplink {
526        Uplink {
527            packets: vec![packet],
528            status: None,
529        }
530    }
531}
532
533impl From<Stat> for Uplink {
534    fn from(status: Stat) -> Uplink {
535        Uplink {
536            packets: Vec::new(),
537            status: Some(status),
538        }
539    }
540}
541
542impl Uplink {
543    /// Writes the payload as the protocol carries it.
544    ///
545    /// # Returns
546    ///
547    /// The JSON text, with an `rxpk` array when there are packets and a `stat` object when
548    /// there is a report.
549    pub fn to_json(&self) -> String {
550        let mut object = Map::new();
551        if !self.packets.is_empty() {
552            object.insert(
553                "rxpk".to_owned(),
554                Value::Array(self.packets.iter().map(Rxpk::to_value).collect()),
555            );
556        }
557        if let Some(status) = &self.status {
558            object.insert("stat".to_owned(), status.to_value());
559        }
560        Value::Object(object).to_string()
561    }
562
563    /// Reads the payload of a PUSH_DATA.
564    ///
565    /// # Arguments
566    ///
567    /// * `body` - the bytes after the header and the gateway's identifier.
568    ///
569    /// # Returns
570    ///
571    /// The packets and the report it carries.
572    ///
573    /// # Errors
574    ///
575    /// Returns [`ProtocolError::Payload`] when the body is not a JSON object, carries neither
576    /// an `rxpk` array nor a `stat` object, or holds an entry the protocol does not describe.
577    pub fn from_json(body: &[u8]) -> Result<Uplink, ProtocolError> {
578        let value = parse_json(body, "a PUSH_DATA")?;
579        let object = object_of(&value, "a PUSH_DATA payload")?;
580        let packets = match object.get("rxpk") {
581            None => Vec::new(),
582            Some(Value::Array(entries)) => entries
583                .iter()
584                .map(Rxpk::from_value)
585                .collect::<Result<Vec<_>, _>>()?,
586            Some(_) => return Err(refused("\"rxpk\" is not an array")),
587        };
588        let status = match object.get("stat") {
589            None => None,
590            Some(value) => Some(Stat::from_value(value)?),
591        };
592        if packets.is_empty() && status.is_none() {
593            return Err(refused(
594                "a PUSH_DATA payload carries neither \"rxpk\" nor \"stat\"",
595            ));
596        }
597        Ok(Uplink { packets, status })
598    }
599}
600
601/// A packet the server asks the gateway to transmit.
602///
603/// # Examples
604///
605/// ```
606/// use pamoja_gateway::udp::Txpk;
607/// use pamoja_lora::LinkSettings;
608///
609/// let downlink = Txpk::at(3_512_348_611, 868_500_000, LinkSettings::new(9, 125_000), b"ok".to_vec())
610///     .with_power_dbm(27)
611///     .with_inverted_polarity(true);
612/// assert!(downlink.to_json().contains("\"ipol\":true"));
613/// ```
614#[derive(Clone, Debug, PartialEq)]
615pub struct Txpk {
616    /// Whether to transmit at once, which ignores the timestamps.
617    pub immediate: bool,
618    /// The concentrator timestamp to transmit at, in microseconds.
619    pub timestamp_us: Option<u32>,
620    /// The GPS time to transmit at, in milliseconds since 6 January 1980.
621    pub gps_millis: Option<u64>,
622    /// The carrier to transmit on, in hertz.
623    pub frequency_hz: u32,
624    /// The radio chain to transmit from.
625    pub rf_chain: u8,
626    /// The power to transmit at, in dBm.
627    pub power_dbm: i8,
628    /// How to modulate it.
629    pub modulation: Modulation,
630    /// The FSK frequency deviation in hertz.
631    pub frequency_deviation_hz: Option<u32>,
632    /// Whether to invert the LoRa polarity, which a LoRaWAN downlink does.
633    pub invert_polarity: bool,
634    /// How long a preamble to send, in symbols.
635    pub preamble_symbols: Option<u16>,
636    /// Whether to leave the physical CRC off, which LoRaWAN downlinks do.
637    pub without_crc: bool,
638    /// The packet itself.
639    pub payload: Vec<u8>,
640}
641
642impl Txpk {
643    /// Describes a LoRa packet to transmit as soon as the gateway can.
644    ///
645    /// # Arguments
646    ///
647    /// * `frequency_hz` - the carrier.
648    /// * `link` - the spreading factor, bandwidth, and coding rate.
649    /// * `payload` - the packet.
650    ///
651    /// # Returns
652    ///
653    /// The request, at 14 dBm on radio chain 0 with standard polarity.
654    pub fn immediate(frequency_hz: u32, link: LinkSettings, payload: Vec<u8>) -> Txpk {
655        Txpk {
656            immediate: true,
657            timestamp_us: None,
658            gps_millis: None,
659            frequency_hz,
660            rf_chain: 0,
661            power_dbm: 14,
662            modulation: Modulation::Lora(link),
663            frequency_deviation_hz: None,
664            invert_polarity: false,
665            preamble_symbols: None,
666            without_crc: false,
667            payload,
668        }
669    }
670
671    /// Describes a LoRa packet to transmit at a concentrator timestamp, which is how a
672    /// LoRaWAN receive window is hit.
673    ///
674    /// # Arguments
675    ///
676    /// * `timestamp_us` - the concentrator counter value to transmit at.
677    /// * `frequency_hz` - the carrier.
678    /// * `link` - the spreading factor, bandwidth, and coding rate.
679    /// * `payload` - the packet.
680    ///
681    /// # Returns
682    ///
683    /// The request.
684    pub fn at(timestamp_us: u32, frequency_hz: u32, link: LinkSettings, payload: Vec<u8>) -> Txpk {
685        Txpk {
686            immediate: false,
687            timestamp_us: Some(timestamp_us),
688            ..Txpk::immediate(frequency_hz, link, payload)
689        }
690    }
691
692    /// Returns it at another power.
693    ///
694    /// # Arguments
695    ///
696    /// * `dbm` - the power in dBm.
697    ///
698    /// # Returns
699    ///
700    /// The request.
701    pub fn with_power_dbm(mut self, dbm: i8) -> Txpk {
702        self.power_dbm = dbm;
703        self
704    }
705
706    /// Returns it with the LoRa polarity inverted or not.
707    ///
708    /// # Arguments
709    ///
710    /// * `inverted` - `true` for a LoRaWAN downlink, which a device listens for inverted.
711    ///
712    /// # Returns
713    ///
714    /// The request.
715    pub fn with_inverted_polarity(mut self, inverted: bool) -> Txpk {
716        self.invert_polarity = inverted;
717        self
718    }
719
720    /// Returns it with the physical CRC left off, as LoRaWAN downlinks are sent.
721    ///
722    /// # Returns
723    ///
724    /// The request.
725    pub fn without_crc(mut self) -> Txpk {
726        self.without_crc = true;
727        self
728    }
729
730    /// Writes the payload of a PULL_RESP.
731    ///
732    /// # Returns
733    ///
734    /// The JSON text, with the request under `txpk`.
735    pub fn to_json(&self) -> String {
736        json!({ "txpk": self.to_value() }).to_string()
737    }
738
739    /// Builds the object the protocol carries.
740    fn to_value(&self) -> Value {
741        let mut object = Map::new();
742        object.insert("imme".to_owned(), json!(self.immediate));
743        if let Some(micros) = self.timestamp_us {
744            object.insert("tmst".to_owned(), json!(micros));
745        }
746        if let Some(millis) = self.gps_millis {
747            object.insert("tmms".to_owned(), json!(millis));
748        }
749        object.insert("freq".to_owned(), megahertz(self.frequency_hz));
750        object.insert("rfch".to_owned(), json!(self.rf_chain));
751        object.insert("powe".to_owned(), json!(self.power_dbm));
752        object.insert("modu".to_owned(), json!(name_of(self.modulation)));
753        object.insert("datr".to_owned(), self.modulation.datarate());
754        match self.modulation {
755            Modulation::Lora(link) => {
756                object.insert(
757                    "codr".to_owned(),
758                    json!(format!("4/{}", link.coding_rate_denominator())),
759                );
760                object.insert("ipol".to_owned(), json!(self.invert_polarity));
761            }
762            Modulation::Fsk(_) => {
763                if let Some(deviation) = self.frequency_deviation_hz {
764                    object.insert("fdev".to_owned(), json!(deviation));
765                }
766            }
767        }
768        if let Some(symbols) = self.preamble_symbols {
769            object.insert("prea".to_owned(), json!(symbols));
770        }
771        object.insert("size".to_owned(), json!(self.payload.len()));
772        object.insert("data".to_owned(), json!(base64::encode(&self.payload)));
773        if self.without_crc {
774            object.insert("ncrc".to_owned(), json!(true));
775        }
776        Value::Object(object)
777    }
778
779    /// Reads the payload of a PULL_RESP.
780    ///
781    /// # Arguments
782    ///
783    /// * `body` - the bytes after the header.
784    ///
785    /// # Returns
786    ///
787    /// The request it carries.
788    ///
789    /// # Errors
790    ///
791    /// Returns [`ProtocolError::Payload`] when the body is not a JSON object with a `txpk`
792    /// the protocol describes.
793    pub fn from_json(body: &[u8]) -> Result<Txpk, ProtocolError> {
794        let value = parse_json(body, "a PULL_RESP")?;
795        let object = object_of(&value, "a PULL_RESP payload")?;
796        let request = object
797            .get("txpk")
798            .ok_or_else(|| refused("a PULL_RESP payload carries no \"txpk\""))?;
799        let request = object_of(request, "a txpk object")?;
800        Ok(Txpk {
801            immediate: flag(request, "imme"),
802            timestamp_us: whole(request, "tmst").map(|value| value as u32),
803            gps_millis: whole(request, "tmms").map(|value| value as u64),
804            frequency_hz: hertz(request, "txpk")?,
805            rf_chain: whole(request, "rfch").unwrap_or(0) as u8,
806            power_dbm: whole(request, "powe").unwrap_or(14) as i8,
807            modulation: modulation_of(request, "txpk")?,
808            frequency_deviation_hz: whole(request, "fdev").map(|value| value as u32),
809            invert_polarity: flag(request, "ipol"),
810            preamble_symbols: whole(request, "prea").map(|value| value as u16),
811            without_crc: flag(request, "ncrc"),
812            payload: payload_of(request, "txpk")?,
813        })
814    }
815}
816
817/// What became of a downlink the server asked for.
818#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
819pub enum TxStatus {
820    /// It was scheduled, which the protocol writes as `NONE`.
821    #[default]
822    None,
823    /// It arrived too late to be scheduled.
824    TooLate,
825    /// Its timestamp is too far ahead.
826    TooEarly,
827    /// Another packet was already scheduled then.
828    CollisionPacket,
829    /// A beacon was already scheduled then.
830    CollisionBeacon,
831    /// The radio chain cannot reach that frequency.
832    TxFreq,
833    /// The gateway cannot transmit at that power.
834    TxPower,
835    /// A GPS timestamp was asked for while the GPS is unlocked.
836    GpsUnlocked,
837}
838
839impl TxStatus {
840    /// Returns the value the protocol writes.
841    ///
842    /// # Returns
843    ///
844    /// The `error` string.
845    pub const fn as_str(self) -> &'static str {
846        match self {
847            TxStatus::None => "NONE",
848            TxStatus::TooLate => "TOO_LATE",
849            TxStatus::TooEarly => "TOO_EARLY",
850            TxStatus::CollisionPacket => "COLLISION_PACKET",
851            TxStatus::CollisionBeacon => "COLLISION_BEACON",
852            TxStatus::TxFreq => "TX_FREQ",
853            TxStatus::TxPower => "TX_POWER",
854            TxStatus::GpsUnlocked => "GPS_UNLOCKED",
855        }
856    }
857
858    /// Names the status a value selects.
859    ///
860    /// # Arguments
861    ///
862    /// * `text` - the `error` string.
863    ///
864    /// # Returns
865    ///
866    /// The status, or `None` for a value the protocol does not define.
867    pub fn named(text: &str) -> Option<TxStatus> {
868        [
869            TxStatus::None,
870            TxStatus::TooLate,
871            TxStatus::TooEarly,
872            TxStatus::CollisionPacket,
873            TxStatus::CollisionBeacon,
874            TxStatus::TxFreq,
875            TxStatus::TxPower,
876            TxStatus::GpsUnlocked,
877        ]
878        .into_iter()
879        .find(|status| status.as_str() == text)
880    }
881
882    /// Reports whether the downlink was scheduled.
883    ///
884    /// # Returns
885    ///
886    /// `true` for [`TxStatus::None`], which is the protocol's way of saying nothing failed.
887    pub const fn scheduled(self) -> bool {
888        matches!(self, TxStatus::None)
889    }
890
891    /// Writes the payload of a TX_ACK.
892    ///
893    /// # Returns
894    ///
895    /// The JSON text, with the status under `txpk_ack`.
896    pub fn to_json(self) -> String {
897        json!({ "txpk_ack": { "error": self.as_str() } }).to_string()
898    }
899
900    /// Reads the payload of a TX_ACK, which a gateway may leave empty when nothing failed.
901    ///
902    /// # Arguments
903    ///
904    /// * `body` - the bytes after the gateway's identifier.
905    ///
906    /// # Returns
907    ///
908    /// The status.
909    ///
910    /// # Errors
911    ///
912    /// Returns [`ProtocolError::Payload`] when the body is neither empty nor a JSON object
913    /// with a `txpk_ack` naming a value the protocol defines.
914    pub fn from_json(body: &[u8]) -> Result<TxStatus, ProtocolError> {
915        if body.iter().all(u8::is_ascii_whitespace) {
916            return Ok(TxStatus::None);
917        }
918        let value = parse_json(body, "a TX_ACK")?;
919        let object = object_of(&value, "a TX_ACK payload")?;
920        let acknowledgment = object
921            .get("txpk_ack")
922            .ok_or_else(|| refused("a TX_ACK payload carries no \"txpk_ack\""))?;
923        let acknowledgment = object_of(acknowledgment, "a txpk_ack object")?;
924        match text(acknowledgment, "error") {
925            None => Ok(TxStatus::None),
926            Some(error) => TxStatus::named(error)
927                .ok_or_else(|| refused(&format!("{error} is not a TX_ACK error"))),
928        }
929    }
930}
931
932impl fmt::Display for TxStatus {
933    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
934        f.write_str(self.as_str())
935    }
936}
937
938/// Names a modulation the way the protocol writes it.
939fn name_of(modulation: Modulation) -> &'static str {
940    match modulation {
941        Modulation::Lora(_) => "LORA",
942        Modulation::Fsk(_) => "FSK",
943    }
944}
945
946/// Writes a frequency in megahertz, which is how the protocol carries one.
947fn megahertz(hertz: u32) -> Value {
948    json!(f64::from(hertz) / 1_000_000.0)
949}
950
951/// Writes a level to the tenth of a decibel the protocol carries.
952fn decibels(value: Decibels) -> Value {
953    json!(f64::from(value.hundredths()) / 100.0)
954}
955
956/// Reads an object's frequency field, in hertz.
957fn hertz(object: &Map<String, Value>, what: &str) -> Result<u32, ProtocolError> {
958    let megahertz =
959        number(object, "freq").ok_or_else(|| refused(&format!("a {what} carries no \"freq\"")))?;
960    let hertz = (megahertz * 1_000_000.0).round();
961    if !(0.0..=f64::from(u32::MAX)).contains(&hertz) {
962        return Err(refused(&format!("{megahertz} MHz is not a carrier")));
963    }
964    Ok(hertz as u32)
965}
966
967/// Reads an object's modulation, from its `modu`, `datr`, and `codr` fields.
968fn modulation_of(object: &Map<String, Value>, what: &str) -> Result<Modulation, ProtocolError> {
969    match text(object, "modu") {
970        Some("FSK") => {
971            let bitrate = whole(object, "datr")
972                .ok_or_else(|| refused(&format!("an FSK {what} carries no bitrate in \"datr\"")))?;
973            Ok(Modulation::Fsk(bitrate as u32))
974        }
975        Some("LORA") | None => {
976            let datarate = text(object, "datr")
977                .ok_or_else(|| refused(&format!("a LoRa {what} carries no \"datr\"")))?;
978            let (factor, bandwidth) = link_of(datarate)
979                .ok_or_else(|| refused(&format!("{datarate} is not a datarate identifier")))?;
980            let mut link = LinkSettings::new(factor, bandwidth);
981            if let Some(coding) = text(object, "codr") {
982                let denominator = coding_rate_of(coding)
983                    .ok_or_else(|| refused(&format!("{coding} is not a coding rate")))?;
984                link = link.with_coding_rate(denominator);
985            }
986            Ok(Modulation::Lora(link))
987        }
988        Some(other) => Err(refused(&format!("{other} is not a modulation"))),
989    }
990}
991
992/// Reads an object's payload, from its base64 `data` field.
993fn payload_of(object: &Map<String, Value>, what: &str) -> Result<Vec<u8>, ProtocolError> {
994    let data =
995        text(object, "data").ok_or_else(|| refused(&format!("a {what} carries no \"data\"")))?;
996    base64::decode(data)
997        .map_err(|error| refused(&format!("the {what} payload is not base64: {error}")))
998}
999
1000/// Parses a datagram's payload as JSON.
1001fn parse_json(body: &[u8], what: &str) -> Result<Value, ProtocolError> {
1002    serde_json::from_slice(body)
1003        .map_err(|error| refused(&format!("the payload of {what} is not JSON: {error}")))
1004}
1005
1006/// Borrows a value as an object.
1007fn object_of<'a>(value: &'a Value, what: &str) -> Result<&'a Map<String, Value>, ProtocolError> {
1008    value
1009        .as_object()
1010        .ok_or_else(|| refused(&format!("{what} is not a JSON object")))
1011}
1012
1013/// Reads a string field.
1014fn text<'a>(object: &'a Map<String, Value>, key: &str) -> Option<&'a str> {
1015    object.get(key).and_then(Value::as_str)
1016}
1017
1018/// Reads a number field.
1019fn number(object: &Map<String, Value>, key: &str) -> Option<f64> {
1020    object.get(key).and_then(Value::as_f64)
1021}
1022
1023/// Reads a whole-number field, which a gateway may write as a float.
1024fn whole(object: &Map<String, Value>, key: &str) -> Option<i64> {
1025    number(object, key).map(|value| value.round() as i64)
1026}
1027
1028/// Reads a flag field, which is absent when it is false.
1029fn flag(object: &Map<String, Value>, key: &str) -> bool {
1030    object.get(key).and_then(Value::as_bool).unwrap_or(false)
1031}
1032
1033/// Builds the error a malformed payload reports.
1034fn refused(why: &str) -> ProtocolError {
1035    ProtocolError::Payload(why.to_owned())
1036}
1037
1038#[cfg(test)]
1039mod tests {
1040    use super::*;
1041
1042    /// The first rxpk entry of the protocol's own example, in section 4.
1043    const EXAMPLE_RXPK: &str = r#"{"rxpk":[{
1044        "time":"2013-03-31T16:21:17.528002Z",
1045        "tmst":3512348611,
1046        "chan":2,
1047        "rfch":0,
1048        "freq":866.349812,
1049        "stat":1,
1050        "modu":"LORA",
1051        "datr":"SF7BW125",
1052        "codr":"4/6",
1053        "rssi":-35,
1054        "lsnr":5.1,
1055        "size":32,
1056        "data":"VEVTVF9QQUNLRVRfMTIzNA=="
1057    }]}"#;
1058
1059    /// The stat example of section 4.
1060    const EXAMPLE_STAT: &str = r#"{"stat":{
1061        "time":"2014-01-12 08:59:28 GMT",
1062        "lati":46.24000,
1063        "long":3.25230,
1064        "alti":145,
1065        "rxnb":2,
1066        "rxok":2,
1067        "rxfw":2,
1068        "ackr":100.0,
1069        "dwnb":2,
1070        "txnb":2
1071    }}"#;
1072
1073    /// The LoRa txpk example of section 6.
1074    const EXAMPLE_TXPK: &str = r#"{"txpk":{
1075        "imme":true,
1076        "freq":864.123456,
1077        "rfch":0,
1078        "powe":14,
1079        "modu":"LORA",
1080        "datr":"SF11BW125",
1081        "codr":"4/6",
1082        "ipol":false,
1083        "size":32,
1084        "data":"H3P3N2i9qc4yt7rK7ldqoeCVJGBybzPY5h1Dd7P7p8v"
1085    }}"#;
1086
1087    #[test]
1088    fn the_protocols_own_rxpk_example_reads() {
1089        let uplink = Uplink::from_json(EXAMPLE_RXPK.as_bytes()).expect("the example parses");
1090        let heard = &uplink.packets[0];
1091
1092        assert_eq!(
1093            heard.received_at.as_deref(),
1094            Some("2013-03-31T16:21:17.528002Z")
1095        );
1096        assert_eq!(heard.timestamp_us, Some(3_512_348_611));
1097        assert_eq!(heard.frequency_hz, 866_349_812);
1098        assert_eq!((heard.channel, heard.rf_chain), (2, 0));
1099        assert_eq!(heard.crc, CrcStatus::Ok);
1100        assert_eq!(
1101            heard.modulation,
1102            Modulation::Lora(LinkSettings::new(7, 125_000).with_coding_rate(6))
1103        );
1104        assert_eq!(heard.rssi_dbm.round_db(), -35);
1105        assert_eq!(heard.snr_db.map(Decibels::hundredths), Some(510));
1106        assert_eq!(heard.payload, b"TEST_PACKET_1234");
1107    }
1108
1109    #[test]
1110    fn the_protocols_own_stat_example_reads() {
1111        let uplink = Uplink::from_json(EXAMPLE_STAT.as_bytes()).expect("the example parses");
1112        let report = uplink.status.expect("it carries a report");
1113
1114        assert_eq!(report.time.as_deref(), Some("2014-01-12 08:59:28 GMT"));
1115        assert_eq!(report.latitude_deg, Some(46.24));
1116        assert_eq!(report.altitude_m, Some(145));
1117        assert_eq!(
1118            (report.received, report.received_ok, report.forwarded),
1119            (2, 2, 2)
1120        );
1121        assert_eq!(report.acknowledged_percent, 100.0);
1122        assert_eq!((report.downlinks, report.transmitted), (2, 2));
1123        assert!(uplink.packets.is_empty());
1124    }
1125
1126    #[test]
1127    fn the_protocols_own_txpk_example_reads() {
1128        let request = Txpk::from_json(EXAMPLE_TXPK.as_bytes()).expect("the example parses");
1129
1130        assert!(request.immediate && !request.invert_polarity);
1131        assert_eq!(request.frequency_hz, 864_123_456);
1132        assert_eq!(request.power_dbm, 14);
1133        assert_eq!(
1134            request.modulation,
1135            Modulation::Lora(LinkSettings::new(11, 125_000).with_coding_rate(6))
1136        );
1137        // The example's payload is written without padding, which the protocol allows.
1138        assert_eq!(request.payload.len(), 32);
1139    }
1140
1141    #[test]
1142    fn an_fsk_packet_carries_its_bitrate_as_a_number() {
1143        let fsk = r#"{"rxpk":[{"tmst":1,"freq":869.1,"stat":1,"modu":"FSK","datr":50000,
1144            "rssi":-75,"size":16,"data":"VEVTVF9QQUNLRVRfMTIzNA=="}]}"#;
1145        let uplink = Uplink::from_json(fsk.as_bytes()).expect("it parses");
1146
1147        assert_eq!(uplink.packets[0].modulation, Modulation::Fsk(50_000));
1148        assert!(uplink.to_json().contains("\"datr\":50000"));
1149    }
1150
1151    #[test]
1152    fn what_is_written_is_read_back() {
1153        let uplink = Uplink {
1154            packets: vec![Rxpk::new(
1155                868_100_000,
1156                LinkSettings::new(12, 125_000).with_coding_rate(8),
1157                b"\x00\x01\xFE\xFF".to_vec(),
1158            )
1159            .with_rssi_dbm(-107)
1160            .with_snr_db(-7.8)
1161            .with_timestamp_us(1_234_567)
1162            .with_received_at(1_364_746_877_528_002)
1163            .on_channel(3, 1)],
1164            status: Some(
1165                Stat::new()
1166                    .at(1_389_517_168)
1167                    .at_position(46.24, 3.2523, 145)
1168                    .with_counts(7, 6, 6)
1169                    .with_downlinks(2, 1)
1170                    .with_acknowledged_percent(99.5),
1171            ),
1172        };
1173
1174        let round = Uplink::from_json(uplink.to_json().as_bytes()).expect("what we wrote parses");
1175        assert_eq!(round, uplink);
1176
1177        let downlink = Txpk::at(
1178            3_512_348_611,
1179            869_525_000,
1180            LinkSettings::new(9, 125_000),
1181            b"downlink".to_vec(),
1182        )
1183        .with_power_dbm(27)
1184        .with_inverted_polarity(true)
1185        .without_crc();
1186        assert_eq!(
1187            Txpk::from_json(downlink.to_json().as_bytes()).expect("what we wrote parses"),
1188            downlink
1189        );
1190    }
1191
1192    #[test]
1193    fn every_tx_ack_value_is_the_one_the_protocol_lists() {
1194        for (status, text) in [
1195            (TxStatus::None, "NONE"),
1196            (TxStatus::TooLate, "TOO_LATE"),
1197            (TxStatus::TooEarly, "TOO_EARLY"),
1198            (TxStatus::CollisionPacket, "COLLISION_PACKET"),
1199            (TxStatus::CollisionBeacon, "COLLISION_BEACON"),
1200            (TxStatus::TxFreq, "TX_FREQ"),
1201            (TxStatus::TxPower, "TX_POWER"),
1202            (TxStatus::GpsUnlocked, "GPS_UNLOCKED"),
1203        ] {
1204            assert_eq!(status.as_str(), text);
1205            assert_eq!(TxStatus::named(text), Some(status));
1206            assert_eq!(TxStatus::from_json(status.to_json().as_bytes()), Ok(status));
1207        }
1208        assert!(TxStatus::None.scheduled() && !TxStatus::TxPower.scheduled());
1209        assert_eq!(TxStatus::from_json(b""), Ok(TxStatus::None));
1210        assert!(TxStatus::named("SOMETHING_ELSE").is_none());
1211    }
1212
1213    #[test]
1214    fn a_payload_the_protocol_does_not_describe_is_refused() {
1215        for (body, why) in [
1216            (r#"{"rxpk":{}}"#, "array"),
1217            (r#"{"rxpk":[{"freq":868.1,"datr":"SF7BW125"}]}"#, "data"),
1218            (r#"{"rxpk":[{"datr":"SF7BW125","data":"aGk="}]}"#, "freq"),
1219            (
1220                r#"{"rxpk":[{"freq":868.1,"datr":"7BW125","data":"aGk="}]}"#,
1221                "datarate",
1222            ),
1223            (
1224                r#"{"rxpk":[{"freq":868.1,"modu":"GFSK","datr":50000,"data":"aGk="}]}"#,
1225                "modulation",
1226            ),
1227            (r#"{}"#, "neither"),
1228        ] {
1229            let refusal = Uplink::from_json(body.as_bytes()).expect_err("it is refused");
1230            assert!(
1231                refusal.to_string().contains(why),
1232                "{body} said {refusal} rather than naming {why}"
1233            );
1234        }
1235    }
1236}
pamoja_gateway/udp/payload.rs

pamoja_gateway/udp/
payload.rs
