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

optimizations

Browse Source
This commit is contained in:
Sergey Savelyev
2024-12-29 20:08:10 -05:00
parent a20db6c522
commit be85ea3aa6
8 changed files with 165 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
Cargo.lock generated
View File

@@ -1,6 +1,6 @@
# This file is automatically @generated by Cargo. # This file is automatically @generated by Cargo.
# It is not intended for manual editing. # It is not intended for manual editing.
version = 3 version = 4
[[package]] [[package]]
name = "actix-codec" name = "actix-codec"
@@ -1146,6 +1146,15 @@ version = "1.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1261fe7e33c73b354eab43b1273a57c8f967d0391e80353e51f764ac02cf6775" checksum = "1261fe7e33c73b354eab43b1273a57c8f967d0391e80353e51f764ac02cf6775"
[[package]]
name = "papaya"
version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ce63bf9dca3eab259cffd421f05661b3386aee36276f5aed9f71450b98f5c5c"
dependencies = [
"seize",
]
[[package]] [[package]]
name = "parking_lot" name = "parking_lot"
version = "0.12.3" version = "0.12.3"
@@ -1445,6 +1454,16 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
[[package]]
name = "seize"
version = "0.4.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d84b0c858bdd30cb56f5597f8b3bf702ec23829e652cc636a1e5a7b9de46ae93"
dependencies = [
"libc",
"windows-sys 0.52.0",
]
[[package]] [[package]]
name = "semver" name = "semver"
version = "1.0.23" version = "1.0.23"
@@ -1507,6 +1526,7 @@ dependencies = [
"futures-util", "futures-util",
"hex", "hex",
"log", "log",
"papaya",
"prost", "prost",
"rand", "rand",
"serde", "serde",

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

@@ -7,6 +7,7 @@ use server::core::{
}; };
use std::error::Error; use std::error::Error;
use std::time::Duration; use std::time::Duration;
use tokio::select;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::sync::mpsc::Sender; use tokio::sync::mpsc::Sender;
use tokio::time::Instant; use tokio::time::Instant;
@@ -38,30 +39,36 @@ impl Telemetry {
let client_stored = client.clone(); let client_stored = client.clone();
let cancel = CancellationToken::new(); let cancel = CancellationToken::new();
let cancel_stored = cancel.clone(); let cancel_stored = cancel.clone();
let (local_tx, mut local_rx) = mpsc::channel(128); let (local_tx, mut local_rx) = mpsc::channel(16);
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(16);
let response_stream = client let response_stream = client
.insert_telemetry(ReceiverStream::new(server_rx)) .insert_telemetry(ReceiverStream::new(server_rx))
.await; .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 { loop {
select! {
_ = cancel.cancelled() => {
break;
},
Some(item) = local_rx.recv() => {
match server_tx.send(item).await { match server_tx.send(item).await {
Ok(_) => {} Ok(_) => {}
Err(_) => break, Err(_) => break,
} }
if let Some(response) = response_stream.next().await { },
Some(response) = response_stream.next() => {
match response { match response {
Ok(_) => {} Ok(_) => {}
Err(_) => { Err(_) => {
break; break;
} }
} }
} else { },
break; else => break,
} }
} }
} else { } else {
@@ -136,6 +143,18 @@ 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 index_handle = tlm
.register("simple_producer/time_offset".into(), TelemetryDataType::Float64)
.await?;
let publish_offset = tlm
.register("simple_producer/publish_offset".into(), TelemetryDataType::Float64)
.await?;
let await_offset = tlm
.register("simple_producer/await_offset".into(), TelemetryDataType::Float64)
.await?;
let sin_tlm_handle = tlm let sin_tlm_handle = tlm
.register("simple_producer/sin".into(), TelemetryDataType::Float32) .register("simple_producer/sin".into(), TelemetryDataType::Float32)
.await?; .await?;
@@ -187,84 +206,98 @@ async fn main() -> Result<(), Box<dyn Error>> {
}); });
} }
let mut next_time = Instant::now(); let start_time = chrono::Utc::now();
let start_instant = Instant::now();
let mut next_time = start_instant;
let mut index = 0; let mut index = 0;
let mut tasks = vec![];
while !cancellation_token.is_cancelled() { while !cancellation_token.is_cancelled() {
next_time += Duration::from_millis(10); next_time += Duration::from_millis(10);
index += 1; index += 1;
tokio::time::sleep_until(next_time).await; tokio::time::sleep_until(next_time).await;
sin_tlm_handle let publish_time = start_time + chrono::TimeDelta::from_std(next_time - start_instant).unwrap();
let actual_time = Instant::now();
tasks.push(index_handle
.publish(
Value::Float64((actual_time - next_time).as_secs_f64()),
chrono::Utc::now(),
));
tasks.push(sin_tlm_handle
.publish( .publish(
Value::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()), Value::Float32((f32::TAU() * (index as f32) / (1000.0_f32)).sin()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(cos_tlm_handle
cos_tlm_handle
.publish( .publish(
Value::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()), Value::Float64((f64::TAU() * (index as f64) / (1000.0_f64)).cos()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(sin2_tlm_handle
sin2_tlm_handle
.publish( .publish(
Value::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()), Value::Float32((f32::TAU() * (index as f32) / (500.0_f32)).sin()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(cos2_tlm_handle
cos2_tlm_handle
.publish( .publish(
Value::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()), Value::Float64((f64::TAU() * (index as f64) / (500.0_f64)).cos()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(sin3_tlm_handle
sin3_tlm_handle
.publish( .publish(
Value::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()), Value::Float32((f32::TAU() * (index as f32) / (333.0_f32)).sin()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(cos3_tlm_handle
cos3_tlm_handle
.publish( .publish(
Value::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()), Value::Float64((f64::TAU() * (index as f64) / (333.0_f64)).cos()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(sin4_tlm_handle
sin4_tlm_handle
.publish( .publish(
Value::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()), Value::Float32((f32::TAU() * (index as f32) / (250.0_f32)).sin()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(cos4_tlm_handle
cos4_tlm_handle
.publish( .publish(
Value::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()), Value::Float64((f64::TAU() * (index as f64) / (250.0_f64)).cos()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(sin5_tlm_handle
sin5_tlm_handle
.publish( .publish(
Value::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()), Value::Float32((f32::TAU() * (index as f32) / (200.0_f32)).sin()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(cos5_tlm_handle
cos5_tlm_handle
.publish( .publish(
Value::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()), Value::Float64((f64::TAU() * (index as f64) / (200.0_f64)).cos()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(sin6_tlm_handle
sin6_tlm_handle
.publish( .publish(
Value::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()), Value::Float32((f32::TAU() * (index as f32) / (166.0_f32)).sin()),
chrono::Utc::now(), publish_time,
) ));
.await?; tasks.push(cos6_tlm_handle
cos6_tlm_handle
.publish( .publish(
Value::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()), Value::Float64((f64::TAU() * (index as f64) / (166.0_f64)).cos()),
publish_time,
));
tasks.push(publish_offset
.publish(
Value::Float64((Instant::now() - actual_time).as_secs_f64()),
chrono::Utc::now(), chrono::Utc::now(),
) ));
.await?;
for task in tasks.drain(..) {
task.await?;
}
tasks.push(await_offset
.publish(
Value::Float64((Instant::now() - actual_time).as_secs_f64()),
chrono::Utc::now(),
));
} }
Ok(()) Ok(())

2
frontend/src/components/GraphAxis.vue
View File

@@ -237,7 +237,7 @@ const lines = computed(() => {
<g class="major_tick" clip-path="url(#y_ticker)"> <g class="major_tick" clip-path="url(#y_ticker)">
<template v-for="tick of lines[1]" :key="tick"> <template v-for="tick of lines[1]" :key="tick">
<polyline <polyline
:points="`${graph_data.x_map(toValue(graph_data.max_x)) - major_tick_length},${y_map(tick)} ${graph_data.x_map(toValue(graph_data.max_x))},${y_map(tick)}`" :points="`${graph_data.x_map(toValue(graph_data.min_x)) + major_tick_length},${y_map(tick)} ${graph_data.x_map(toValue(graph_data.min_x))},${y_map(tick)}`"
></polyline> ></polyline>
<text <text
class="left_edge middle_text" class="left_edge middle_text"

2
frontend/src/components/TelemetryLine.vue
View File

@@ -163,7 +163,7 @@ watch(
return ''; return '';
} }
const future_number = toValue(graph_data.max_x) + 9999999999; const future_number = toValue(graph_data.max_x) + 99999999;
let last_x = graph_data.x_map(future_number); let last_x = graph_data.x_map(future_number);
old_max.value = toValue(graph_data.max_x); old_max.value = toValue(graph_data.max_x);

54
frontend/src/views/HomeView.vue
View File

@@ -4,6 +4,7 @@ import { provide } from 'vue';
import Graph from '@/components/SvgGraph.vue'; import Graph from '@/components/SvgGraph.vue';
import Axis from '@/components/GraphAxis.vue'; import Axis from '@/components/GraphAxis.vue';
import Line from '@/components/TelemetryLine.vue'; import Line from '@/components/TelemetryLine.vue';
import { AxisSide } from '@/graph/axis';
const websocket = useWebsocket(); const websocket = useWebsocket();
provide(WEBSOCKET_SYMBOL, websocket); provide(WEBSOCKET_SYMBOL, websocket);
@@ -11,6 +12,18 @@ provide(WEBSOCKET_SYMBOL, websocket);
<template> <template>
<main> <main>
<Graph
:width="800"
:height="400"
:border_top_bottom="24"
:border_left_right="128"
>
<Axis>
<Line data="simple_producer/time_offset"></Line>
<Line data="simple_producer/publish_offset"></Line>
<Line data="simple_producer/await_offset"></Line>
</Axis>
</Graph>
<Graph <Graph
:width="800" :width="800"
:height="400" :height="400"
@@ -20,41 +33,28 @@ provide(WEBSOCKET_SYMBOL, websocket);
<Axis> <Axis>
<Line data="simple_producer/sin"></Line> <Line data="simple_producer/sin"></Line>
<Line data="simple_producer/cos4"></Line> <Line data="simple_producer/cos4"></Line>
</Axis>
</Graph>
<Graph
:width="800"
:height="400"
:border_top_bottom="24"
:border_left_right="128"
>
<Axis>
<Line data="simple_producer/sin2"></Line> <Line data="simple_producer/sin2"></Line>
<Line data="simple_producer/cos"></Line> <Line data="simple_producer/cos"></Line>
</Axis>
</Graph>
<Graph
:width="800"
:height="400"
:border_top_bottom="24"
:border_left_right="128"
>
<Axis>
<Line data="simple_producer/sin3"></Line> <Line data="simple_producer/sin3"></Line>
<Line data="simple_producer/cos2"></Line> <Line data="simple_producer/cos2"></Line>
</Axis>
</Graph>
<Graph
:width="800"
:height="400"
:border_top_bottom="24"
:border_left_right="128"
>
<Axis>
<Line data="simple_producer/sin4"></Line> <Line data="simple_producer/sin4"></Line>
<Line data="simple_producer/cos3"></Line> <Line data="simple_producer/cos3"></Line>
</Axis> </Axis>
</Graph> </Graph>
<Graph
:width="800"
:height="400"
:border_top_bottom="24"
:border_left_right="128"
>
</Graph>
<Graph
:width="800"
:height="400"
:border_top_bottom="24"
:border_left_right="128"
>
</Graph>
</main> </main>
</template> </template>

1
server/Cargo.toml
View File

@@ -21,6 +21,7 @@ serde_json = "1.0.132"
derive_more = { version = "1.0.0", features = ["full"] } derive_more = { version = "1.0.0", features = ["full"] }
hex = "0.4.3" hex = "0.4.3"
futures-util = "0.3.31" futures-util = "0.3.31"
papaya = "0.1.7"
[build-dependencies] [build-dependencies]
tonic-build = "0.12.3" tonic-build = "0.12.3"

5
server/src/http.rs
View File

@@ -113,7 +113,10 @@ async fn websocket_connect(
break; break;
} }
Ok(_) = rx.changed() => { Ok(_) = rx.changed() => {
let value = rx.borrow_and_update().clone(); let value = {
let ref_val = rx.borrow_and_update();
ref_val.clone()
};
let _ = tx.send(WebsocketResponse::TlmValue { let _ = tx.send(WebsocketResponse::TlmValue {
uuid: uuid.clone(), uuid: uuid.clone(),
value, value,

29
server/src/telemetry.rs
View File

@@ -6,7 +6,7 @@ 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 tokio::sync::Mutex; use tokio::sync::RwLock;
fn tlm_data_type_serialzier<S>( fn tlm_data_type_serialzier<S>(
tlm_data_type: &TelemetryDataType, tlm_data_type: &TelemetryDataType,
@@ -70,15 +70,15 @@ pub struct TelemetryData {
} }
pub struct TelemetryManagementService { pub struct TelemetryManagementService {
uuid_mapping: Arc<Mutex<HashMap<String, String>>>, uuid_mapping: Arc<RwLock<HashMap<String, String>>>,
tlm_mapping: Arc<Mutex<HashMap<String, TelemetryData>>>, tlm_mapping: Arc<RwLock<HashMap<String, TelemetryData>>>,
} }
impl TelemetryManagementService { impl TelemetryManagementService {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
uuid_mapping: Arc::new(Mutex::new(HashMap::new())), uuid_mapping: Arc::new(RwLock::new(HashMap::new())),
tlm_mapping: Arc::new(Mutex::new(HashMap::new())), tlm_mapping: Arc::new(RwLock::new(HashMap::new())),
} }
} }
@@ -86,10 +86,10 @@ impl TelemetryManagementService {
&self, &self,
telemetry_definition_request: TelemetryDefinitionRequest, telemetry_definition_request: TelemetryDefinitionRequest,
) -> Result<String, Box<dyn Error>> { ) -> Result<String, Box<dyn Error>> {
let mut lock = self.uuid_mapping.lock().await; let lock = self.uuid_mapping.read().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.read().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());
@@ -103,7 +103,9 @@ impl TelemetryManagementService {
"Adding New Telemetry Definition {:?}", "Adding New Telemetry Definition {:?}",
telemetry_definition_request telemetry_definition_request
); );
let mut tlm_lock = self.tlm_mapping.lock().await; drop(lock);
let mut lock = self.uuid_mapping.write().await;
let mut tlm_lock = self.tlm_mapping.write().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( tlm_lock.insert(
@@ -122,16 +124,13 @@ impl TelemetryManagementService {
} }
pub async fn get_by_name(&self, name: &String) -> Option<TelemetryData> { pub async fn get_by_name(&self, name: &String) -> Option<TelemetryData> {
let uuid = { let uuid_lock = self.uuid_mapping.read().await;
let uuid_lock = self.uuid_mapping.lock().await; let uuid = uuid_lock.get(name)?;
uuid_lock.get(name).cloned()? self.get_by_uuid(uuid).await
};
self.get_by_uuid(&uuid).await
} }
pub async fn get_by_uuid(&self, uuid: &String) -> Option<TelemetryData> { pub async fn get_by_uuid(&self, uuid: &String) -> Option<TelemetryData> {
let tlm_lock = self.tlm_mapping.lock().await; let tlm_lock = self.tlm_mapping.read().await;
tlm_lock.get(uuid).cloned() tlm_lock.get(uuid).cloned()
} }
} }