fix: add proper epoch serialization
This commit is contained in:
parent
dc81c21ea9
commit
6090b2b47f
8 changed files with 95 additions and 35 deletions
|
@ -6,11 +6,10 @@
|
||||||
// #[serde(rename = "...")]
|
// #[serde(rename = "...")]
|
||||||
// #[serde(alias = "...")]
|
// #[serde(alias = "...")]
|
||||||
|
|
||||||
use crate::utils::SupportedUnit;
|
use crate::utils::{EpochUTC, SupportedUnit};
|
||||||
use crate::ingest_protocol::error::Error;
|
use crate::ingest_protocol::error::Error;
|
||||||
use crate::ingest_protocol::parser::parse_mac_address;
|
use crate::ingest_protocol::parser::parse_mac_address;
|
||||||
|
|
||||||
use hifitime::Epoch;
|
|
||||||
use rust_decimal::Decimal;
|
use rust_decimal::Decimal;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use serde_with::serde_as;
|
use serde_with::serde_as;
|
||||||
|
@ -33,7 +32,7 @@ use std::hash::{Hash, Hasher};
|
||||||
pub struct SensorValue {
|
pub struct SensorValue {
|
||||||
pub mac: String,
|
pub mac: String,
|
||||||
pub value: Decimal,
|
pub value: Decimal,
|
||||||
pub time: Option<Epoch>,
|
pub time: Option<EpochUTC>,
|
||||||
pub unit: Option<SupportedUnit>,
|
pub unit: Option<SupportedUnit>,
|
||||||
pub name: Option<String>,
|
pub name: Option<String>,
|
||||||
}
|
}
|
||||||
|
@ -59,7 +58,7 @@ serde_with::serde_conv!(
|
||||||
);
|
);
|
||||||
|
|
||||||
#[serde_as]
|
#[serde_as]
|
||||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)]
|
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
pub struct NMDeviceDataPacket {
|
pub struct NMDeviceDataPacket {
|
||||||
#[serde(alias = "ID")]
|
#[serde(alias = "ID")]
|
||||||
#[serde_as(as = "MacAsArray")]
|
#[serde_as(as = "MacAsArray")]
|
||||||
|
@ -80,7 +79,7 @@ pub struct NMDeviceDataPacket {
|
||||||
/// - Если в values есть хотябы один times и этот time не None => Игнорируем этот time
|
/// - Если в values есть хотябы один times и этот time не None => Игнорируем этот time
|
||||||
/// - Если time нет у values => используем этот time (если он None, то соотв. считаем что информации о времени нет)
|
/// - Если time нет у values => используем этот time (если он None, то соотв. считаем что информации о времени нет)
|
||||||
/// TODO: В базе всё должно храниться как секунды по TAI спустя J1900 для возможности сортировки по времени позже.
|
/// TODO: В базе всё должно храниться как секунды по TAI спустя J1900 для возможности сортировки по времени позже.
|
||||||
pub time: Option<Epoch>,
|
pub time: Option<EpochUTC>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Default, PartialEq, Deserialize)]
|
#[derive(Debug, Clone, Default, PartialEq, Deserialize)]
|
||||||
|
|
|
@ -94,7 +94,7 @@ pub fn parse_packet_body<'a>(
|
||||||
packet.values.insert(SensorValue {
|
packet.values.insert(SensorValue {
|
||||||
mac: sensor_mac.into(),
|
mac: sensor_mac.into(),
|
||||||
value: Decimal::from_str(sensor_value)?,
|
value: Decimal::from_str(sensor_value)?,
|
||||||
time: sensor_time, // TODO
|
time: sensor_time.map(|v| v.into()), // TODO
|
||||||
unit: None,
|
unit: None,
|
||||||
name: sensor_name.map(|v| v.to_string()),
|
name: sensor_name.map(|v| v.to_string()),
|
||||||
});
|
});
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
use hifitime::Epoch;
|
||||||
use crate::ingest_protocol::parser::{parse_mac_address, parse_packet};
|
use crate::ingest_protocol::parser::{parse_mac_address, parse_packet};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
@ -5,7 +6,7 @@ fn test_asd() {
|
||||||
let asd = r#"
|
let asd = r#"
|
||||||
#A2-C6-47-01-DF-E1#Метео
|
#A2-C6-47-01-DF-E1#Метео
|
||||||
#OWNER#unknown
|
#OWNER#unknown
|
||||||
#T1#13.44#Outdoor
|
#T1#13.44#1000000#Outdoor
|
||||||
#T2#27.74#Indoor
|
#T2#27.74#Indoor
|
||||||
#P1#691.02#Barometer
|
#P1#691.02#Barometer
|
||||||
#LAT#55.738178
|
#LAT#55.738178
|
||||||
|
@ -14,14 +15,18 @@ fn test_asd() {
|
||||||
##
|
##
|
||||||
"#
|
"#
|
||||||
.trim();
|
.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]
|
#[test]
|
||||||
fn test_mac() {
|
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]
|
#[test]
|
||||||
|
|
|
@ -3,9 +3,10 @@
|
||||||
#![feature(try_blocks)]
|
#![feature(try_blocks)]
|
||||||
extern crate core;
|
extern crate core;
|
||||||
|
|
||||||
mod utils;
|
|
||||||
mod ingest_protocol;
|
mod ingest_protocol;
|
||||||
mod web_server;
|
mod web_server;
|
||||||
|
mod utils;
|
||||||
|
|
||||||
use crate::web_server::server_main;
|
use crate::web_server::server_main;
|
||||||
|
|
||||||
struct Params {}
|
struct Params {}
|
||||||
|
|
64
src/utils/hifitime_serde.rs
Normal file
64
src/utils/hifitime_serde.rs
Normal file
|
@ -0,0 +1,64 @@
|
||||||
|
use std::fmt;
|
||||||
|
use std::fmt::{Formatter, Write};
|
||||||
|
use hifitime::Epoch;
|
||||||
|
use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
|
||||||
|
|
||||||
|
#[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 or a float")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn visit_i64<E>(self, v: i64) -> Result<Self::Value, E> where E: de::Error {
|
||||||
|
return Ok(Epoch::from_unix_seconds(v as f64).into())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn visit_f32<E>(self, v: f32) -> 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<EpochUTC, E>
|
||||||
|
where
|
||||||
|
E: de::Error,
|
||||||
|
{
|
||||||
|
return Ok(Epoch::from_unix_seconds(
|
||||||
|
value.parse().map_err(de::Error::custom)?
|
||||||
|
).into())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
deserializer.deserialize_any(EpochVisitor)
|
||||||
|
}
|
||||||
|
}
|
|
@ -2,10 +2,14 @@
|
||||||
//!
|
//!
|
||||||
|
|
||||||
|
|
||||||
|
mod hifitime_serde;
|
||||||
|
|
||||||
use phf::phf_map;
|
use phf::phf_map;
|
||||||
use serde::{Deserialize, Deserializer, Serialize, Serializer};
|
use serde::{Deserialize, Deserializer, Serialize, Serializer};
|
||||||
use std::borrow::Cow;
|
use std::borrow::Cow;
|
||||||
|
|
||||||
|
pub use hifitime_serde::EpochUTC;
|
||||||
|
|
||||||
|
|
||||||
/// Поддерживаемые типы.
|
/// Поддерживаемые типы.
|
||||||
///
|
///
|
||||||
|
@ -31,8 +35,8 @@ pub enum SupportedUnit {
|
||||||
|
|
||||||
impl Serialize for SupportedUnit {
|
impl Serialize for SupportedUnit {
|
||||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||||
where
|
where
|
||||||
S: Serializer,
|
S: Serializer,
|
||||||
{
|
{
|
||||||
serializer.serialize_str(match self {
|
serializer.serialize_str(match self {
|
||||||
SupportedUnit::Celsius => "C",
|
SupportedUnit::Celsius => "C",
|
||||||
|
@ -54,8 +58,8 @@ impl Serialize for SupportedUnit {
|
||||||
|
|
||||||
impl<'de> Deserialize<'de> for SupportedUnit {
|
impl<'de> Deserialize<'de> for SupportedUnit {
|
||||||
fn deserialize<'a, D>(deserializer: D) -> Result<Self, D::Error>
|
fn deserialize<'a, D>(deserializer: D) -> Result<Self, D::Error>
|
||||||
where
|
where
|
||||||
D: Deserializer<'de>,
|
D: Deserializer<'de>,
|
||||||
{
|
{
|
||||||
let color_str = Cow::<'a, str>::deserialize(deserializer)?;
|
let color_str = Cow::<'a, str>::deserialize(deserializer)?;
|
||||||
match STR_TO_UNITS.get(color_str.as_ref()) {
|
match STR_TO_UNITS.get(color_str.as_ref()) {
|
||||||
|
@ -78,5 +82,5 @@ static STR_TO_UNITS: phf::Map<&'static str, SupportedUnit> = phf_map! {
|
||||||
"V" => SupportedUnit::Volts,
|
"V" => SupportedUnit::Volts,
|
||||||
"W" => SupportedUnit::Watts,
|
"W" => SupportedUnit::Watts,
|
||||||
"s" => SupportedUnit::Seconds,
|
"s" => SupportedUnit::Seconds,
|
||||||
"KWh" => SupportedUnit::Seconds,
|
"KWh" => SupportedUnit::KWh,
|
||||||
};
|
};
|
|
@ -6,9 +6,8 @@ use crate::ingest_protocol::{NMDeviceDataPacket, NMJsonPacket};
|
||||||
use crate::web_server::app_error::AppError;
|
use crate::web_server::app_error::AppError;
|
||||||
|
|
||||||
|
|
||||||
use fred::types::RedisMap;
|
|
||||||
use ntex::http::{HttpMessage, StatusCode};
|
use ntex::http::{HttpMessage, StatusCode};
|
||||||
use ntex::util::{Bytes, HashMap};
|
use ntex::util::Bytes;
|
||||||
use ntex::{http, web};
|
use ntex::{http, web};
|
||||||
use fred::prelude::*;
|
use fred::prelude::*;
|
||||||
use hifitime::Epoch;
|
use hifitime::Epoch;
|
||||||
|
@ -80,7 +79,7 @@ pub async fn device_handler<'a>(
|
||||||
for device in real_body.devices {
|
for device in real_body.devices {
|
||||||
let mut device_key_str = String::new();
|
let mut device_key_str = String::new();
|
||||||
|
|
||||||
let now = Epoch::now().unwrap();
|
let now = Epoch::now().unwrap().into();
|
||||||
let mut device_time = device.time
|
let mut device_time = device.time
|
||||||
.unwrap_or(now);
|
.unwrap_or(now);
|
||||||
|
|
||||||
|
@ -90,7 +89,7 @@ pub async fn device_handler<'a>(
|
||||||
device_time = now;
|
device_time = now;
|
||||||
}
|
}
|
||||||
|
|
||||||
let device_tai_timestamp = device_time
|
let device_tai_timestamp = device_time.0
|
||||||
.to_duration_since_j1900()
|
.to_duration_since_j1900()
|
||||||
.to_seconds();
|
.to_seconds();
|
||||||
|
|
||||||
|
|
|
@ -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> {
|
pub async fn parse_nm_qs_format(input: &str) -> Result<NMDeviceDataPacket, QSParserError> {
|
||||||
let mut parsed: HashMap<String, String> = serde_qs::from_str(input)?;
|
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)?
|
parse_mac_address(id)?
|
||||||
} else {
|
} else {
|
||||||
return Err(QSParserError::NoMAC);
|
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 {
|
let device_data = NMDeviceDataPacket {
|
||||||
mac: _device_mac,
|
mac: device_mac,
|
||||||
name: parsed.remove("name").map(|v| v.to_owned()),
|
name: parsed.remove("name").map(|v| v.to_owned()),
|
||||||
owner: parsed.remove("owner").map(|v| v.to_owned()),
|
owner: parsed.remove("owner").map(|v| v.to_owned()),
|
||||||
lat: parse_decimal_if_exists(&mut parsed, "lat")?,
|
lat: parse_decimal_if_exists(&mut parsed, "lat")?,
|
||||||
lon: parse_decimal_if_exists(&mut parsed, "lon")?,
|
lon: parse_decimal_if_exists(&mut parsed, "lon")?,
|
||||||
alt: parse_decimal_if_exists(&mut parsed, "alt")?,
|
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)?,
|
values: qs_rest_to_values(parsed)?,
|
||||||
};
|
};
|
||||||
|
|
Loading…
Add table
Reference in a new issue