sergeysav/telemetry_visualization
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixes formatting and lints

Browse Source
This commit is contained in:
Sergey Savelyev
2024-12-27 11:05:47 -05:00
parent 147d1abaf8
commit 12f27bad69
11 changed files with 281 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

227
examples/simple_producer/src/main.rs
View File

@@ -1,7 +1,10 @@
use chrono::DateTime; use chrono::DateTime;
use num_traits::float::FloatConst;
use server::core::telemetry_service_client::TelemetryServiceClient; use server::core::telemetry_service_client::TelemetryServiceClient;
use server::core::telemetry_value::Value; use server::core::telemetry_value::Value;
use server::core::{TelemetryDataType, TelemetryDefinitionRequest, TelemetryItem, TelemetryValue, Timestamp, Uuid}; use server::core::{
TelemetryDataType, TelemetryDefinitionRequest, TelemetryItem, TelemetryValue, Timestamp, Uuid,
};
use std::error::Error; use std::error::Error;
use std::time::Duration; use std::time::Duration;
use tokio::sync::mpsc; use tokio::sync::mpsc;
@@ -13,7 +16,6 @@ use tonic::codegen::tokio_stream::StreamExt;
use tonic::codegen::StdError; use tonic::codegen::StdError;
use tonic::transport::Channel; use tonic::transport::Channel;
use tonic::Request; use tonic::Request;
use num_traits::float::FloatConst;
struct Telemetry { struct Telemetry {
client: TelemetryServiceClient<Channel>, client: TelemetryServiceClient<Channel>,
@@ -41,22 +43,22 @@ impl Telemetry {
tokio::spawn(async move { tokio::spawn(async move {
while !cancel.is_cancelled() { while !cancel.is_cancelled() {
let (server_tx, server_rx) = mpsc::channel(1); let (server_tx, server_rx) = mpsc::channel(1);
let response_stream = client.insert_telemetry(ReceiverStream::new(server_rx)).await; let response_stream = client
.insert_telemetry(ReceiverStream::new(server_rx))
.await;
if let Ok(response_stream) = response_stream { if let Ok(response_stream) = response_stream {
let mut response_stream = response_stream.into_inner(); let mut response_stream = response_stream.into_inner();
while let Some(item) = local_rx.recv().await { while let Some(item) = local_rx.recv().await {
match server_tx.send(item).await { match server_tx.send(item).await {
Ok(_) => {}, Ok(_) => {}
Err(_) => { Err(_) => break,
break
},
} }
if let Some(response) = response_stream.next().await { if let Some(response) = response_stream.next().await {
match response { match response {
Ok(_) => {}, Ok(_) => {}
Err(_) => { Err(_) => {
break; break;
}, }
} }
} else { } else {
break; break;
@@ -75,13 +77,23 @@ impl Telemetry {
}) })
} }
pub async fn register(&mut self, name: String, data_type: TelemetryDataType) -> Result<TelemetryItemHandle, Box<dyn Error>> { pub async fn register(
let response = self.client.new_telemetry(Request::new(TelemetryDefinitionRequest { &mut self,
name: String,
data_type: TelemetryDataType,
) -> Result<TelemetryItemHandle, Box<dyn Error>> {
let response = self
.client
.new_telemetry(Request::new(TelemetryDefinitionRequest {
name, name,
data_type: data_type.into(), data_type: data_type.into(),
})).await?.into_inner(); }))
.await?
.into_inner();
let Some(uuid) = response.uuid else { return Err("UUID Missing".into()); }; let Some(uuid) = response.uuid else {
return Err("UUID Missing".into());
};
Ok(TelemetryItemHandle { Ok(TelemetryItemHandle {
uuid: uuid.value, uuid: uuid.value,
@@ -97,20 +109,25 @@ impl Drop for Telemetry {
} }
impl TelemetryItemHandle { impl TelemetryItemHandle {
pub async fn publish(&self, value: Value, timestamp: DateTime<chrono::Utc>) -> Result<(), Box<dyn Error>> { pub async fn publish(
let offset_from_unix_epoch = timestamp - DateTime::from_timestamp(0, 0).expect("Could not get Unix epoch"); &self,
self.tx.send(TelemetryItem { value: Value,
timestamp: DateTime<chrono::Utc>,
) -> Result<(), Box<dyn Error>> {
let offset_from_unix_epoch =
timestamp - DateTime::from_timestamp(0, 0).expect("Could not get Unix epoch");
self.tx
.send(TelemetryItem {
uuid: Some(Uuid { uuid: Some(Uuid {
value: self.uuid.clone() value: self.uuid.clone(),
}),
value: Some(TelemetryValue {
value: Some(value)
}), }),
value: Some(TelemetryValue { value: Some(value) }),
timestamp: Some(Timestamp { timestamp: Some(Timestamp {
secs: offset_from_unix_epoch.num_seconds(), secs: offset_from_unix_epoch.num_seconds(),
nanos: offset_from_unix_epoch.subsec_nanos(), nanos: offset_from_unix_epoch.subsec_nanos(),
}), }),
}).await?; })
.await?;
Ok(()) Ok(())
} }
} }
@@ -119,23 +136,47 @@ impl TelemetryItemHandle {
async fn main() -> Result<(), Box<dyn Error>> { async fn main() -> Result<(), Box<dyn Error>> {
let mut tlm = Telemetry::new("http://[::1]:50051").await?; let mut tlm = Telemetry::new("http://[::1]:50051").await?;
let sin_tlm_handle = tlm.register("simple_producer/sin".into(), TelemetryDataType::Float32).await?; let sin_tlm_handle = tlm
let cos_tlm_handle = tlm.register("simple_producer/cos".into(), TelemetryDataType::Float64).await?; .register("simple_producer/sin".into(), TelemetryDataType::Float32)
.await?;
let cos_tlm_handle = tlm
.register("simple_producer/cos".into(), TelemetryDataType::Float64)
.await?;
let sin2_tlm_handle = tlm.register("simple_producer/sin2".into(), TelemetryDataType::Float32).await?; let sin2_tlm_handle = tlm
let cos2_tlm_handle = tlm.register("simple_producer/cos2".into(), TelemetryDataType::Float64).await?; .register("simple_producer/sin2".into(), TelemetryDataType::Float32)
.await?;
let cos2_tlm_handle = tlm
.register("simple_producer/cos2".into(), TelemetryDataType::Float64)
.await?;
let sin3_tlm_handle = tlm.register("simple_producer/sin3".into(), TelemetryDataType::Float32).await?; let sin3_tlm_handle = tlm
let cos3_tlm_handle = tlm.register("simple_producer/cos3".into(), TelemetryDataType::Float64).await?; .register("simple_producer/sin3".into(), TelemetryDataType::Float32)
.await?;
let cos3_tlm_handle = tlm
.register("simple_producer/cos3".into(), TelemetryDataType::Float64)
.await?;
let sin4_tlm_handle = tlm.register("simple_producer/sin4".into(), TelemetryDataType::Float32).await?; let sin4_tlm_handle = tlm
let cos4_tlm_handle = tlm.register("simple_producer/cos4".into(), TelemetryDataType::Float64).await?; .register("simple_producer/sin4".into(), TelemetryDataType::Float32)
.await?;
let cos4_tlm_handle = tlm
.register("simple_producer/cos4".into(), TelemetryDataType::Float64)
.await?;
let sin5_tlm_handle = tlm.register("simple_producer/sin5".into(), TelemetryDataType::Float32).await?; let sin5_tlm_handle = tlm
let cos5_tlm_handle = tlm.register("simple_producer/cos5".into(), TelemetryDataType::Float64).await?; .register("simple_producer/sin5".into(), TelemetryDataType::Float32)
.await?;
let cos5_tlm_handle = tlm
.register("simple_producer/cos5".into(), TelemetryDataType::Float64)
.await?;
let sin6_tlm_handle = tlm.register("simple_producer/sin6".into(), TelemetryDataType::Float32).await?; let sin6_tlm_handle = tlm
let cos6_tlm_handle = tlm.register("simple_producer/cos6".into(), TelemetryDataType::Float64).await?; .register("simple_producer/sin6".into(), TelemetryDataType::Float32)
.await?;
let cos6_tlm_handle = tlm
.register("simple_producer/cos6".into(), TelemetryDataType::Float64)
.await?;
let cancellation_token = CancellationToken::new(); let cancellation_token = CancellationToken::new();
{ {
@@ -152,78 +193,78 @@ async fn main() -> Result<(), Box<dyn Error>> {
next_time += Duration::from_millis(100); next_time += Duration::from_millis(100);
index += 10; index += 10;
tokio::time::sleep_until(next_time).await; tokio::time::sleep_until(next_time).await;
sin_tlm_handle.publish( sin_tlm_handle
Value::Float32( .publish(
(f32::TAU() * (index as f32) / (1000.0_f32)).sin() Value::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()),
),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
cos_tlm_handle.publish( .await?;
Value::Float64( cos_tlm_handle
(f64::TAU() * (index as f64) / (1000.0_f64)).cos() .publish(
), Value::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
sin2_tlm_handle.publish( .await?;
Value::Float32( sin2_tlm_handle
(f32::TAU() * (index as f32) / (500.0_f32)).sin() .publish(
), Value::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
cos2_tlm_handle.publish( .await?;
Value::Float64( cos2_tlm_handle
(f64::TAU() * (index as f64) / (500.0_f64)).cos() .publish(
), Value::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
sin3_tlm_handle.publish( .await?;
Value::Float32( sin3_tlm_handle
(f32::TAU() * (index as f32) / (333.0_f32)).sin() .publish(
), Value::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
cos3_tlm_handle.publish( .await?;
Value::Float64( cos3_tlm_handle
(f64::TAU() * (index as f64) / (333.0_f64)).cos() .publish(
), Value::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
sin4_tlm_handle.publish( .await?;
Value::Float32( sin4_tlm_handle
(f32::TAU() * (index as f32) / (250.0_f32)).sin() .publish(
), Value::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
cos4_tlm_handle.publish( .await?;
Value::Float64( cos4_tlm_handle
(f64::TAU() * (index as f64) / (250.0_f64)).cos() .publish(
), Value::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
sin5_tlm_handle.publish( .await?;
Value::Float32( sin5_tlm_handle
(f32::TAU() * (index as f32) / (200.0_f32)).sin() .publish(
), Value::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
cos5_tlm_handle.publish( .await?;
Value::Float64( cos5_tlm_handle
(f64::TAU() * (index as f64) / (200.0_f64)).cos() .publish(
), Value::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
sin6_tlm_handle.publish( .await?;
Value::Float32( sin6_tlm_handle
(f32::TAU() * (index as f32) / (166.0_f32)).sin() .publish(
), Value::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
cos6_tlm_handle.publish( .await?;
Value::Float64( cos6_tlm_handle
(f64::TAU() * (index as f64) / (166.0_f64)).cos() .publish(
), Value::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()),
chrono::Utc::now(), chrono::Utc::now(),
).await?; )
.await?;
} }
Ok(()) Ok(())

3
frontend/src/components/SvgGraph.vue
View File

@@ -140,8 +140,7 @@ const lines = computed(() => {
:timestamp="tick" :timestamp="tick"
:utc="props.utc" :utc="props.utc"
:show_millis="duration < 1000" :show_millis="duration < 1000"
> ></TimeText>
</TimeText>
</template> </template>
</g> </g>
<slot></slot> <slot></slot>

14
frontend/src/components/TimeText.vue
View File

@@ -5,8 +5,8 @@ const props = defineProps<{
x: number; x: number;
y: number; y: number;
timestamp: number; timestamp: number;
utc?: boolean utc?: boolean;
show_millis?: boolean show_millis?: boolean;
}>(); }>();
// This function is slow // This function is slow
@@ -61,18 +61,12 @@ function getDateString(date: Date) {
const timetext = computed(() => { const timetext = computed(() => {
return getDateString(new Date(props.timestamp)); return getDateString(new Date(props.timestamp));
}); });
</script> </script>
<template> <template>
<text <text :x="props.x" :y="props.y">
:x="props.x"
:y="props.y"
>
{{ timetext }} {{ timetext }}
</text> </text>
</template> </template>
<style scoped lang="scss"> <style scoped lang="scss"></style>
</style>

20
frontend/src/components/ValueLabel.vue
View File

@@ -1,6 +1,5 @@
<script setup lang="ts"> <script setup lang="ts">
import { computed, ref, useTemplateRef, watch, watchPostEffect } from 'vue'; import { computed, ref, useTemplateRef, watch } from 'vue';
import { useNow } from '@/composables/ticker';
const props = defineProps<{ const props = defineProps<{
x: number; x: number;
@@ -19,12 +18,15 @@ const value_text = computed(() => {
const label_width = ref(0); const label_width = ref(0);
watch([value_text, labelRef], ([_, label]) => { watch(
[value_text, labelRef],
([, label]) => {
label_width.value = label?.getBBox().width || 0; label_width.value = label?.getBBox().width || 0;
}, { },
{
flush: 'post', flush: 'post',
}); },
);
</script> </script>
<template> <template>
@@ -34,11 +36,7 @@ watch([value_text, labelRef], ([_, label]) => {
:width="label_width + background_offset * 2" :width="label_width + background_offset * 2"
:height="16 + background_offset * 2" :height="16 + background_offset * 2"
></rect> ></rect>
<text <text ref="label-ref" :x="x" :y="y">
ref="label-ref"
:x="x"
:y="y"
>
{{ value_text }} {{ value_text }}
</text> </text>
</template> </template>

1
server/build.rs
View File

@@ -1,4 +1,3 @@
fn main() -> Result<(), Box<dyn std::error::Error>> { fn main() -> Result<(), Box<dyn std::error::Error>> {
tonic_build::compile_protos("proto/core.proto")?; tonic_build::compile_protos("proto/core.proto")?;
Ok(()) Ok(())

65
server/src/grpc.rs
View File

@@ -1,20 +1,23 @@
use crate::core::telemetry_service_server::{TelemetryService, TelemetryServiceServer};
use crate::core::telemetry_value::Value;
use crate::core::{
TelemetryDataType, TelemetryDefinitionRequest, TelemetryDefinitionResponse,
TelemetryInsertResponse, TelemetryItem, TelemetryValue, Uuid,
};
use crate::telemetry::{TelemetryDataItem, TelemetryDataValue, TelemetryManagementService};
use chrono::{DateTime, SecondsFormat};
use log::{error, trace};
use std::error::Error; use std::error::Error;
use std::pin::Pin; use std::pin::Pin;
use std::sync::Arc; use std::sync::Arc;
use chrono::{DateTime, SecondsFormat};
use log::{error, trace};
use tokio::select; use tokio::select;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use tonic::{Request, Response, Status, Streaming};
use tonic::codegen::tokio_stream::{Stream, StreamExt};
use tonic::codegen::tokio_stream::wrappers::ReceiverStream; use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
use tonic::codegen::tokio_stream::{Stream, StreamExt};
use tonic::transport::Server; use tonic::transport::Server;
use crate::core::telemetry_service_server::{TelemetryService, TelemetryServiceServer}; use tonic::{Request, Response, Status, Streaming};
use crate::core::{TelemetryDataType, TelemetryDefinitionRequest, TelemetryDefinitionResponse, TelemetryInsertResponse, TelemetryItem, TelemetryValue, Uuid};
use crate::core::telemetry_value::Value;
use crate::telemetry::{TelemetryDataItem, TelemetryDataValue, TelemetryManagementService};
pub struct CoreTelemetryService { pub struct CoreTelemetryService {
pub tlm_management: Arc<TelemetryManagementService>, pub tlm_management: Arc<TelemetryManagementService>,
@@ -23,24 +26,29 @@ pub struct CoreTelemetryService {
#[tonic::async_trait] #[tonic::async_trait]
impl TelemetryService for CoreTelemetryService { impl TelemetryService for CoreTelemetryService {
async fn new_telemetry(&self, request: Request<TelemetryDefinitionRequest>) -> Result<Response<TelemetryDefinitionResponse>, Status> { async fn new_telemetry(
&self,
request: Request<TelemetryDefinitionRequest>,
) -> Result<Response<TelemetryDefinitionResponse>, Status> {
trace!("CoreTelemetryService::new_telemetry"); trace!("CoreTelemetryService::new_telemetry");
self.tlm_management.register(request.into_inner()).await self.tlm_management
.map(|uuid| .register(request.into_inner())
.await
.map(|uuid| {
Response::new(TelemetryDefinitionResponse { Response::new(TelemetryDefinitionResponse {
uuid: Some(Uuid { uuid: Some(Uuid { value: uuid }),
value: uuid
}),
}) })
) })
.map_err(|err| .map_err(|err| Status::already_exists(err.to_string()))
Status::already_exists(err.to_string())
)
} }
type InsertTelemetryStream = Pin<Box<dyn Stream<Item = Result<TelemetryInsertResponse, Status>> + Send>>; type InsertTelemetryStream =
Pin<Box<dyn Stream<Item = Result<TelemetryInsertResponse, Status>> + Send>>;
async fn insert_telemetry(&self, request: Request<Streaming<TelemetryItem>>) -> Result<Response<Self::InsertTelemetryStream>, Status> { async fn insert_telemetry(
&self,
request: Request<Streaming<TelemetryItem>>,
) -> Result<Response<Self::InsertTelemetryStream>, Status> {
trace!("CoreTelemetryService::insert_telemetry"); trace!("CoreTelemetryService::insert_telemetry");
let cancel_token = self.cancellation_token.clone(); let cancel_token = self.cancellation_token.clone();
@@ -77,7 +85,10 @@ impl TelemetryService for CoreTelemetryService {
} }
impl CoreTelemetryService { impl CoreTelemetryService {
async fn handle_new_tlm_item(tlm_management: &Arc<TelemetryManagementService>, tlm_item: &TelemetryItem) -> Result<TelemetryInsertResponse, Status> { async fn handle_new_tlm_item(
tlm_management: &Arc<TelemetryManagementService>,
tlm_item: &TelemetryItem,
) -> Result<TelemetryInsertResponse, Status> {
trace!("CoreTelemetryService::handle_new_tlm_item {:?}", tlm_item); trace!("CoreTelemetryService::handle_new_tlm_item {:?}", tlm_item);
let Some(ref uuid) = tlm_item.uuid else { let Some(ref uuid) = tlm_item.uuid else {
return Err(Status::failed_precondition("UUID Missing")); return Err(Status::failed_precondition("UUID Missing"));
@@ -102,7 +113,8 @@ impl CoreTelemetryService {
return Err(Status::failed_precondition("Data Type Mismatch")); return Err(Status::failed_precondition("Data Type Mismatch"));
}; };
let Some(timestamp) = DateTime::from_timestamp(timestamp.secs, timestamp.nanos as u32) else { let Some(timestamp) = DateTime::from_timestamp(timestamp.secs, timestamp.nanos as u32)
else {
return Err(Status::invalid_argument("Failed to construct UTC DateTime")); return Err(Status::invalid_argument("Failed to construct UTC DateTime"));
}; };
@@ -111,19 +123,22 @@ impl CoreTelemetryService {
Value::Float32(x) => TelemetryDataValue::Float32(x), Value::Float32(x) => TelemetryDataValue::Float32(x),
Value::Float64(x) => TelemetryDataValue::Float64(x), Value::Float64(x) => TelemetryDataValue::Float64(x),
}, },
timestamp: timestamp.to_rfc3339_opts(SecondsFormat::Millis, true) timestamp: timestamp.to_rfc3339_opts(SecondsFormat::Millis, true),
})); }));
Ok(TelemetryInsertResponse {}) Ok(TelemetryInsertResponse {})
} }
} }
pub fn setup(token: CancellationToken, telemetry_management_service: Arc<TelemetryManagementService>) -> Result<JoinHandle<()>, Box<dyn Error>> { pub fn setup(
token: CancellationToken,
telemetry_management_service: Arc<TelemetryManagementService>,
) -> Result<JoinHandle<()>, Box<dyn Error>> {
let addr = "[::1]:50051".parse()?; let addr = "[::1]:50051".parse()?;
Ok(tokio::spawn(async move { Ok(tokio::spawn(async move {
let tlm_service = CoreTelemetryService { let tlm_service = CoreTelemetryService {
tlm_management: telemetry_management_service, tlm_management: telemetry_management_service,
cancellation_token: token.clone() cancellation_token: token.clone(),
}; };
trace!("Starting gRPC Server"); trace!("Starting gRPC Server");

49
server/src/http.rs
View File

@@ -1,15 +1,15 @@
use crate::telemetry::{TelemetryDataItem, TelemetryManagementService};
use actix_web::http::header::ContentType; use actix_web::http::header::ContentType;
use actix_web::http::StatusCode; use actix_web::http::StatusCode;
use actix_web::{rt, error, get, web, App, HttpRequest, HttpResponse, HttpServer, Responder}; use actix_web::{error, get, rt, web, App, HttpRequest, HttpResponse, HttpServer, Responder};
use derive_more::{Display, Error};
use std::sync::Arc;
use actix_ws::AggregatedMessage; use actix_ws::AggregatedMessage;
use derive_more::{Display, Error};
use log::{error, trace}; use log::{error, trace};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::select; use tokio::select;
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use tonic::codegen::tokio_stream::StreamExt; use tonic::codegen::tokio_stream::StreamExt;
use crate::telemetry::{TelemetryDataItem, TelemetryManagementService};
#[derive(Debug, Display, Error)] #[derive(Debug, Display, Error)]
enum UserError { enum UserError {
@@ -18,35 +18,36 @@ enum UserError {
} }
impl error::ResponseError for UserError { impl error::ResponseError for UserError {
fn error_response(&self) -> HttpResponse {
HttpResponse::build(self.status_code())
.insert_header(ContentType::html())
.body(self.to_string())
}
fn status_code(&self) -> StatusCode { fn status_code(&self) -> StatusCode {
match *self { match *self {
UserError::TlmNotFound { .. } => StatusCode::NOT_FOUND, UserError::TlmNotFound { .. } => StatusCode::NOT_FOUND,
} }
} }
fn error_response(&self) -> HttpResponse {
HttpResponse::build(self.status_code())
.insert_header(ContentType::html())
.body(self.to_string())
}
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
enum WebsocketRequest { enum WebsocketRequest {
RegisterTlmListener { RegisterTlmListener { uuid: String },
uuid: String
}
} }
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
enum WebsocketResponse { enum WebsocketResponse {
TlmValue { TlmValue {
uuid: String, uuid: String,
value: Option<TelemetryDataItem> value: Option<TelemetryDataItem>,
} },
} }
#[get("/tlm/{name:[\\w\\d/_-]+}")] #[get("/tlm/{name:[\\w\\d/_-]+}")]
async fn get_tlm_definition(data: web::Data<Arc<TelemetryManagementService>>, name: web::Path<String>) -> Result<impl Responder, UserError> { async fn get_tlm_definition(
data: web::Data<Arc<TelemetryManagementService>>,
name: web::Path<String>,
) -> Result<impl Responder, UserError> {
let string = name.to_string(); let string = name.to_string();
trace!("get_tlm_definition {}", string); trace!("get_tlm_definition {}", string);
let Some(data) = data.get_by_name(&string).await else { let Some(data) = data.get_by_name(&string).await else {
@@ -56,7 +57,12 @@ async fn get_tlm_definition(data: web::Data<Arc<TelemetryManagementService>>, na
Ok(web::Json(data.definition.clone())) Ok(web::Json(data.definition.clone()))
} }
async fn websocket_connect(req: HttpRequest, stream: web::Payload, data: web::Data<Arc<TelemetryManagementService>>, cancel_token: web::Data<CancellationToken>) -> Result<HttpResponse, actix_web::Error> { async fn websocket_connect(
req: HttpRequest,
stream: web::Payload,
data: web::Data<Arc<TelemetryManagementService>>,
cancel_token: web::Data<CancellationToken>,
) -> Result<HttpResponse, actix_web::Error> {
trace!("websocket_connect"); trace!("websocket_connect");
let (res, mut session, stream) = actix_ws::handle(&req, stream)?; let (res, mut session, stream) = actix_ws::handle(&req, stream)?;
@@ -157,11 +163,13 @@ async fn websocket_connect(req: HttpRequest, stream: web::Payload, data: web::Da
} }
fn setup_api(cfg: &mut web::ServiceConfig) { fn setup_api(cfg: &mut web::ServiceConfig) {
cfg cfg.service(get_tlm_definition);
.service(get_tlm_definition);
} }
pub async fn setup(cancellation_token: CancellationToken, telemetry_definitions: Arc<TelemetryManagementService>) -> Result<(), Box<dyn Error>> { pub async fn setup(
cancellation_token: CancellationToken,
telemetry_definitions: Arc<TelemetryManagementService>,
) -> Result<(), Box<dyn Error>> {
let data = web::Data::new(telemetry_definitions); let data = web::Data::new(telemetry_definitions);
let cancel_token = web::Data::new(cancellation_token); let cancel_token = web::Data::new(cancellation_token);
@@ -174,7 +182,8 @@ pub async fn setup(cancellation_token: CancellationToken, telemetry_definitions:
.service(web::scope("/api").configure(setup_api)) .service(web::scope("/api").configure(setup_api))
}) })
.bind("localhost:8080")? .bind("localhost:8080")?
.run().await?; .run()
.await?;
Ok(()) Ok(())
} }

3
server/src/lib.rs
View File

@@ -1,7 +1,7 @@
mod uuid;
mod grpc; mod grpc;
mod http; mod http;
mod telemetry; mod telemetry;
mod uuid;
pub mod core { pub mod core {
tonic::include_proto!("core"); tonic::include_proto!("core");
@@ -33,4 +33,3 @@ pub async fn setup() -> Result<(), Box<dyn Error>> {
Ok(()) Ok(())
} }

5
server/src/main.rs
View File

@@ -5,9 +5,8 @@ use std::str::FromStr;
async fn main() -> Result<(), Box<dyn std::error::Error>> { async fn main() -> Result<(), Box<dyn std::error::Error>> {
let log_file = env::var("LOG_FILE"); let log_file = env::var("LOG_FILE");
let log_level = match env::var("LOG_LEVEL") { let log_level = match env::var("LOG_LEVEL") {
Ok(log_level) => log::LevelFilter::from_str(&log_level) Ok(log_level) => log::LevelFilter::from_str(&log_level).unwrap_or(log::LevelFilter::Info),
.unwrap_or(log::LevelFilter::Info), Err(_) => log::LevelFilter::Info,
Err(_) => log::LevelFilter::Info
}; };
let mut log_config = fern::Dispatch::new() let mut log_config = fern::Dispatch::new()

44
server/src/telemetry.rs
View File

@@ -1,14 +1,20 @@
use crate::core::{TelemetryDataType, TelemetryDefinitionRequest, Uuid};
use log::trace;
use serde::de::Visitor;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::collections::HashMap; use std::collections::HashMap;
use std::error::Error; use std::error::Error;
use std::fmt::Formatter; use std::fmt::Formatter;
use std::sync::Arc; use std::sync::Arc;
use log::trace;
use serde::de::Visitor;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use tokio::sync::Mutex; use tokio::sync::Mutex;
use crate::core::{TelemetryDataType, TelemetryDefinitionRequest, Uuid};
fn tlm_data_type_serialzier<S>(tlm_data_type: &TelemetryDataType, serializer: S) -> Result<S::Ok, S::Error> where S: Serializer { fn tlm_data_type_serialzier<S>(
tlm_data_type: &TelemetryDataType,
serializer: S,
) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(tlm_data_type.as_str_name()) serializer.serialize_str(tlm_data_type.as_str_name())
} }
@@ -29,7 +35,10 @@ impl Visitor<'_> for TlmDataTypeVisitor {
} }
} }
fn tlm_data_type_deserialzier<'de, D>(deserializer: D) -> Result<TelemetryDataType, D::Error> where D: Deserializer<'de> { fn tlm_data_type_deserialzier<'de, D>(deserializer: D) -> Result<TelemetryDataType, D::Error>
where
D: Deserializer<'de>,
{
deserializer.deserialize_str(TlmDataTypeVisitor) deserializer.deserialize_str(TlmDataTypeVisitor)
} }
@@ -51,7 +60,7 @@ pub enum TelemetryDataValue {
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TelemetryDataItem { pub struct TelemetryDataItem {
pub value: TelemetryDataValue, pub value: TelemetryDataValue,
pub timestamp: String pub timestamp: String,
} }
#[derive(Clone)] #[derive(Clone)]
@@ -73,12 +82,15 @@ impl TelemetryManagementService {
} }
} }
pub async fn register(&self, telemetry_definition_request: TelemetryDefinitionRequest) -> Result<String, Box<dyn Error>> { pub async fn register(
&self,
telemetry_definition_request: TelemetryDefinitionRequest,
) -> Result<String, Box<dyn Error>> {
let mut lock = self.uuid_mapping.lock().await; let mut lock = self.uuid_mapping.lock().await;
if let Some(uuid) = lock.get(&telemetry_definition_request.name) { if let Some(uuid) = lock.get(&telemetry_definition_request.name) {
trace!("Telemetry Definition Found {:?}", uuid); trace!("Telemetry Definition Found {:?}", uuid);
let tlm_lock = self.tlm_mapping.lock().await; let tlm_lock = self.tlm_mapping.lock().await;
if let Some(TelemetryData { definition, ..}) = tlm_lock.get(uuid) { if let Some(TelemetryData { definition, .. }) = tlm_lock.get(uuid) {
if definition.data_type != telemetry_definition_request.data_type() { if definition.data_type != telemetry_definition_request.data_type() {
return Err("A telemetry item of the same name already exists".into()); return Err("A telemetry item of the same name already exists".into());
} }
@@ -87,18 +99,24 @@ impl TelemetryManagementService {
Err("Could not find Telemetry Data".into()) Err("Could not find Telemetry Data".into())
} }
} else { } else {
trace!("Adding New Telemetry Definition {:?}", telemetry_definition_request); trace!(
"Adding New Telemetry Definition {:?}",
telemetry_definition_request
);
let mut tlm_lock = self.tlm_mapping.lock().await; let mut tlm_lock = self.tlm_mapping.lock().await;
let uuid = Uuid::random().value; let uuid = Uuid::random().value;
lock.insert(telemetry_definition_request.name.clone(), uuid.clone()); lock.insert(telemetry_definition_request.name.clone(), uuid.clone());
tlm_lock.insert(uuid.clone(), TelemetryData { tlm_lock.insert(
uuid.clone(),
TelemetryData {
definition: TelemetryDefinition { definition: TelemetryDefinition {
uuid: uuid.clone(), uuid: uuid.clone(),
name: telemetry_definition_request.name.clone(), name: telemetry_definition_request.name.clone(),
data_type: telemetry_definition_request.data_type(), data_type: telemetry_definition_request.data_type(),
}, },
data: tokio::sync::watch::channel(None).0 data: tokio::sync::watch::channel(None).0,
}); },
);
Ok(uuid) Ok(uuid)
} }
} }

4
server/src/uuid.rs
View File

@@ -1,9 +1,9 @@
use crate::core::Uuid;
use rand::RngCore; use rand::RngCore;
use crate::core::{Uuid};
impl Uuid { impl Uuid {
pub fn random() -> Self { pub fn random() -> Self {
let mut uuid =[0u8; 16]; let mut uuid = [0u8; 16];
rand::thread_rng().fill_bytes(&mut uuid); rand::thread_rng().fill_bytes(&mut uuid);
Self { Self {
value: hex::encode(uuid), value: hex::encode(uuid),