Merge pull request 'Исправление сериализации Epoch' (#13) from fix/hifitime-serde into master

Reviewed-on: nm17/iotishnik-server#13
Reviewed-by: DarkCat09 <gh@dc09.ru>

src/ingest_protocol/packet_types.rs

@@ -6,11 +6,10 @@
 // #[serde(rename = "...")]
 // #[serde(alias = "...")]
 
-use crate::utils::SupportedUnit;
+use crate::utils::{EpochUTC, SupportedUnit};
 use crate::ingest_protocol::error::Error;
 use crate::ingest_protocol::parser::parse_mac_address;
 
-use hifitime::Epoch;
 use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
@@ -33,7 +32,7 @@ use std::hash::{Hash, Hasher};
 pub struct SensorValue {
     pub mac: String,
     pub value: Decimal,
-    pub time: Option<Epoch>,
+    pub time: Option<EpochUTC>,
     pub unit: Option<SupportedUnit>,
     pub name: Option<String>,
 }
@@ -59,7 +58,7 @@ serde_with::serde_conv!(
 );
 
 #[serde_as]
-#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)]
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub struct NMDeviceDataPacket {
     #[serde(alias = "ID")]
     #[serde_as(as = "MacAsArray")]
@@ -80,7 +79,7 @@ pub struct NMDeviceDataPacket {
     ///   - Если в values есть хотябы один times и этот time не None => Игнорируем этот time
     ///   - Если time нет у values => используем этот time (если он None, то соотв. считаем что информации о времени нет)
     /// TODO: В базе всё должно храниться как секунды по TAI спустя J1900 для возможности сортировки по времени позже.
-    pub time: Option<Epoch>,
+    pub time: Option<EpochUTC>,
 }
 
 #[derive(Debug, Clone, Default, PartialEq, Deserialize)]

src/ingest_protocol/parser.rs

@@ -94,7 +94,7 @@ pub fn parse_packet_body<'a>(
                 packet.values.insert(SensorValue {
                     mac: sensor_mac.into(),
                     value: Decimal::from_str(sensor_value)?,
-                    time: sensor_time, // TODO
+                    time: sensor_time.map(|v| v.into()), // TODO
                     unit: None,
                     name: sensor_name.map(|v| v.to_string()),
                 });

src/ingest_protocol/tests/mod.rs

@@ -1,11 +1,15 @@
-use crate::ingest_protocol::parser::{parse_mac_address, parse_packet};
+use crate::ingest_protocol::{
+    parser::{parse_mac_address, parse_packet},
+    SensorValue,
+};
+use hifitime::Epoch;
 
 #[test]
 fn test_asd() {
     let asd = r#"
 #A2-C6-47-01-DF-E1#Метео
 #OWNER#unknown
-#T1#13.44#Outdoor
+#T1#13.44#1000000#Outdoor
 #T2#27.74#Indoor
 #P1#691.02#Barometer
 #LAT#55.738178
@@ -14,14 +18,22 @@ fn test_asd() {
 ##
 "#
     .trim();
-    dbg!(parse_packet(asd));
+    let packet = parse_packet(asd).unwrap().1;
+    assert!(serde_json::to_string_pretty(&packet)
+        .unwrap()
+        .contains(r#""time": 1000000.0"#));
 }
 
 #[test]
 fn test_mac() {
-    //assert_eq!(parse_mac_address("12-34-AA-12-55-AA"), Ok(("", [18, 52, 170, 18, 85, 170])) );
+    assert_eq!(
+        parse_mac_address("12-34-AA-12-55-AA").unwrap(),
+        ("", [18, 52, 170, 18, 85, 170])
+    );
 
-    println!("{:?}", parse_mac_address("12-34-AA-12-55-AA"));
+    dbg!(Epoch::now().unwrap().to_unix_seconds());
+
+    //println!("{:?}", parse_mac_address("12-34-AA-12-55-AA"));
 }
 
 #[test]
@@ -35,3 +47,42 @@ fn test_packet() {
 
     println!("{:#?}", parse_packet(inp));
 }
+
+#[test]
+fn test_hex_parsing() {
+    let inp = r#"
+        {"mac": "T1", "value": "123", "time": "665B514B"}
+    "#
+    .trim();
+
+    let sensor_value: SensorValue = serde_json::from_str(inp).unwrap();
+    assert_eq!(
+        sensor_value.time.unwrap().0.to_unix_seconds(),
+        1717260619.0,
+        "hex text"
+    );
+
+    let inp = r#"
+        {"mac": "T1", "value": "123", "time": 1717260619}
+    "#
+    .trim();
+
+    let sensor_value: SensorValue = serde_json::from_str(inp).unwrap();
+    assert_eq!(
+        sensor_value.time.unwrap().0.to_unix_seconds(),
+        1717260619.0,
+        "integer test"
+    );
+
+    let inp = r#"
+        {"mac": "T1", "value": "123", "time": 1717260619.0}
+    "#
+    .trim();
+
+    let sensor_value: SensorValue = serde_json::from_str(inp).unwrap();
+    assert_eq!(
+        sensor_value.time.unwrap().0.to_unix_seconds(),
+        1717260619.0,
+        "float test"
+    );
+}

src/main.rs

@@ -3,9 +3,10 @@
 #![feature(try_blocks)]
 extern crate core;
 
-mod utils;
 mod ingest_protocol;
 mod web_server;
+mod utils;
+
 use crate::web_server::server_main;
 
 struct Params {}

src/utils/hifitime_serde.rs

@@ -0,0 +1,76 @@
+use hifitime::Epoch;
+use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
+use std::fmt;
+use std::fmt::{Formatter, Write};
+
+#[derive(PartialOrd, PartialEq, Ord, Eq, Clone, Copy, Debug, Default)]
+#[repr(transparent)]
+pub struct EpochUTC(pub Epoch);
+
+impl From<Epoch> for EpochUTC {
+    #[inline]
+    fn from(value: Epoch) -> Self {
+        EpochUTC(value)
+    }
+}
+
+impl Serialize for EpochUTC {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_f64(self.0.to_unix_seconds())
+    }
+}
+
+impl<'de> Deserialize<'de> for EpochUTC {
+    fn deserialize<D>(deserializer: D) -> Result<EpochUTC, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        struct EpochVisitor;
+
+        impl<'de> de::Visitor<'de> for EpochVisitor {
+            type Value = EpochUTC;
+
+            fn expecting(&self, formatter: &mut Formatter) -> fmt::Result {
+                formatter.write_str(
+                    "a (string, representing a timestamp in decimal or hexadecimal form) or a decimal",
+                )
+            }
+
+            fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+            where
+                E: de::Error,
+            {
+                return Ok(Epoch::from_unix_seconds(v as f64).into());
+            }
+
+            fn visit_f64<E>(self, v: f64) -> Result<Self::Value, E>
+            where
+                E: de::Error,
+            {
+                return Ok(Epoch::from_unix_seconds(v).into());
+            }
+
+            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
+            where
+                E: de::Error,
+            {
+                if value.len() == 8 {
+                    let bytes = hex::decode(value).map_err(de::Error::custom)?;
+
+                    return Ok(Epoch::from_unix_seconds(u32::from_be_bytes(
+                        bytes.as_slice().try_into().unwrap(),
+                    ) as f64)
+                    .into());
+                }
+                return Ok(
+                    Epoch::from_unix_seconds(value.parse().map_err(de::Error::custom)?).into(),
+                );
+            }
+        }
+
+        deserializer.deserialize_any(EpochVisitor)
+    }
+}

src/utils/mod.rs

@@ -2,10 +2,14 @@
 //!
 
 
+mod hifitime_serde;
+
 use phf::phf_map;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::borrow::Cow;
 
+pub use hifitime_serde::EpochUTC;
+
 
 /// Поддерживаемые типы.
 ///
@@ -78,5 +82,5 @@ static STR_TO_UNITS: phf::Map<&'static str, SupportedUnit> = phf_map! {
     "V" => SupportedUnit::Volts,
     "W" => SupportedUnit::Watts,
     "s" => SupportedUnit::Seconds,
-    "KWh" => SupportedUnit::Seconds,
+    "KWh" => SupportedUnit::KWh,
 };

src/web_server/old_device_sensor_api/mod.rs

@@ -6,9 +6,8 @@ use crate::ingest_protocol::{NMDeviceDataPacket, NMJsonPacket};
 use crate::web_server::app_error::AppError;
 
 
-use fred::types::RedisMap;
 use ntex::http::{HttpMessage, StatusCode};
-use ntex::util::{Bytes, HashMap};
+use ntex::util::Bytes;
 use ntex::{http, web};
 use fred::prelude::*;
 use hifitime::Epoch;
@@ -80,7 +79,7 @@ pub async fn device_handler<'a>(
         for device in real_body.devices {
             let mut device_key_str = String::new();
 
-            let now = Epoch::now().unwrap();
+            let now = Epoch::now().unwrap().into();
             let mut device_time = device.time
                 .unwrap_or(now);
 
@@ -90,7 +89,7 @@ pub async fn device_handler<'a>(
                 device_time = now;
             }
 
-            let device_tai_timestamp = device_time
+            let device_tai_timestamp = device_time.0
                 .to_duration_since_j1900()
                 .to_seconds();
 

src/web_server/old_device_sensor_api/qs_parser.rs

@@ -85,32 +85,20 @@ pub fn parse_epoch_if_exists(parsed: &mut HashMap<String, String>, key: &str) ->
 pub async fn parse_nm_qs_format(input: &str) -> Result<NMDeviceDataPacket, QSParserError> {
     let mut parsed: HashMap<String, String> = serde_qs::from_str(input)?;
 
-    let (_, _device_mac) = if let Some(id) = parsed.get("ID") {
+    let (_, device_mac) = if let Some(id) = parsed.get("ID") {
         parse_mac_address(id)?
     } else {
         return Err(QSParserError::NoMAC);
     };
 
-    for keys in parsed.keys() {
-        if keys.starts_with("") {
-
-        }
-    }
-
-    let lat = if let Some(lat) = parsed.remove("lat") {
-        Some(Decimal::from_str(lat.as_str())?)
-    } else {
-        None
-    };
-
     let device_data = NMDeviceDataPacket {
-        mac: _device_mac,
+        mac: device_mac,
         name: parsed.remove("name").map(|v| v.to_owned()),
         owner: parsed.remove("owner").map(|v| v.to_owned()),
         lat: parse_decimal_if_exists(&mut parsed, "lat")?,
         lon: parse_decimal_if_exists(&mut parsed, "lon")?,
         alt: parse_decimal_if_exists(&mut parsed, "alt")?,
-        time: parse_epoch_if_exists(&mut parsed, "time")?,
+        time: parse_epoch_if_exists(&mut parsed, "time")?.map(|v| v.into()),
 
         values: qs_rest_to_values(parsed)?,
     };