Skip to content
This repository was archived by the owner on Nov 1, 2024. It is now read-only.

added performance metrics LabGraph issue 78 #88

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

added performance metrics LabGraph issue 78 #88

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,4 +42,5 @@ venv.bak/

# Sphinx
sphinx/_build/
*~
*~
node_modules
38 changes: 35 additions & 3 deletions extensions/graphviz_support/graphviz_support/tests/demo_graph/amplifier.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,16 @@
import labgraph as lg
import numpy as np
import time
import random
from .random_message import RandomMessage
from datetime import datetime
from time import perf_counter
from collections import deque


data = deque()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the future, deque can be replaced by a more performant buffer data structure.

throughput_data = ""
datarate_data = ""
one_second_must_pass = perf_counter()
class AmplifierConfig(lg.Config):
out_in_ratio: float

Expand All @@ -22,10 +29,35 @@ def output(self, _in: float) -> float:
@lg.subscriber(AMPLIFIER_INPUT)
@lg.publisher(AMPLIFIER_OUTPUT)
async def amplify(self, message: RandomMessage) -> lg.AsyncPublisher:
current_time = time.time()
global data
global throughput_data
global datarate_data
global one_second_must_pass
Copy link
Contributor

@jfResearchEng jfResearchEng Aug 18, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please try to avoid using global here.


current_time_UTC = str(datetime.utcnow()) + " (UTC)"
start_time = perf_counter()
# timestamp_data = time.time()
output_data = np.array(
[self.output(_in) for _in in message.data]
)


data.append(message.data.size * message.data.itemsize)
if perf_counter() - one_second_must_pass >= 1:
tempThroughput = 0
tempDataRate = 0
for messagedata in data:
tempThroughput += messagedata
tempDataRate += 1.0
throughput_data = str(tempThroughput) + " bytes"
datarate_data = str(tempDataRate) + " msg/sec"
data.clear()
one_second_must_pass = perf_counter()

latency_data = str(perf_counter() - start_time) + " s"
yield self.AMPLIFIER_OUTPUT, RandomMessage(
timestamp=current_time, data=output_data
timestamp=current_time_UTC, data=output_data, latency=latency_data, throughput=throughput_data, datarate=datarate_data
)



39 changes: 36 additions & 3 deletions extensions/graphviz_support/graphviz_support/tests/demo_graph/attenuator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,23 @@
import labgraph as lg
import numpy as np
import time
import random
from .random_message import RandomMessage
from datetime import datetime
from time import perf_counter
from collections import deque

data = deque()
throughput_data = ""
datarate_data = ""
one_second_must_pass = perf_counter()

class AttenuatorConfig(lg.Config):
attenuation: float


class CustomMessage(RandomMessage):
latency: float

class Attenuator(lg.Node):
ATTENUATOR_INPUT = lg.Topic(RandomMessage)
ATTENUATOR_OUTPUT = lg.Topic(RandomMessage)
Expand All @@ -22,10 +32,33 @@ def output(self, _in: float) -> float:
@lg.subscriber(ATTENUATOR_INPUT)
@lg.publisher(ATTENUATOR_OUTPUT)
async def attenuate(self, message: RandomMessage) -> lg.AsyncPublisher:
current_time = time.time()
global data
global throughput_data
global datarate_data
global one_second_must_pass
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here for global.


start_time = perf_counter()
current_time_UCT = str(datetime.utcnow()) + " (UTC)"
output_data = np.array(
[self.output(_in) for _in in message.data]
)

data.append(message.data.size * message.data.itemsize)
if perf_counter() - one_second_must_pass >= 1:
tempThroughput = 0
tempDataRate = 0
for messagedata in data:
tempThroughput += messagedata
tempDataRate += 1.0
throughput_data = str(tempThroughput) + " bytes"
datarate_data = str(tempDataRate) + " msg/sec"
data.clear()
one_second_must_pass = perf_counter()

latency_data = str(perf_counter() - start_time) + " s"
yield self.ATTENUATOR_OUTPUT, RandomMessage(
timestamp=current_time, data=output_data
timestamp=current_time_UCT, data=output_data, latency=latency_data,throughput=throughput_data, datarate=datarate_data
)



23 changes: 19 additions & 4 deletions extensions/graphviz_support/graphviz_support/tests/demo_graph/noise_generator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,23 +5,38 @@
import numpy as np
import asyncio
import time
import random
from .random_message import RandomMessage

from datetime import datetime
from time import perf_counter

class NoiseGeneratorConfig(lg.Config):
sample_rate: float
num_features: int


class NoiseGenerator(lg.Node):
NOISE_GENERATOR_OUTPUT = lg.Topic(RandomMessage)
config: NoiseGeneratorConfig

@lg.publisher(NOISE_GENERATOR_OUTPUT)
async def generate_noise(self) -> lg.AsyncPublisher:
while True:
start_time = perf_counter()
timestamp_data = time.time()

datarate_data = random.random() + 6
throughput_data = len(str(np.random.rand(self.config.num_features)))
latency_data = perf_counter() - start_time

yield self.NOISE_GENERATOR_OUTPUT, RandomMessage(
timestamp=time.time(),
data=np.random.rand(self.config.num_features)
timestamp=timestamp_data,
data=np.random.rand(self.config.num_features),

latency=10.0,
throughput=throughput_data,
datarate=datarate_data
)
await asyncio.sleep(1 / self.config.sample_rate)


46 changes: 41 additions & 5 deletions extensions/graphviz_support/graphviz_support/tests/demo_graph/rolling_averager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,18 @@
import labgraph as lg
import numpy as np
import time
import random
from typing import List
from dataclasses import field
from .random_message import RandomMessage
from datetime import datetime
from time import perf_counter
from collections import deque

data = deque()
throughput_data = ""
datarate_data = ""
one_second_must_pass = perf_counter()

class RollingState(lg.State):
messages: List[RandomMessage] = field(default_factory=list)
Expand All @@ -16,31 +24,59 @@ class RollingState(lg.State):
class RollingConfig(lg.Config):
window: float


class RollingAverager(lg.Node):
ROLLING_AVERAGER_INPUT = lg.Topic(RandomMessage)
ROLLING_AVERAGER_OUTPUT = lg.Topic(RandomMessage)

state: RollingState
config: RollingConfig



@lg.subscriber(ROLLING_AVERAGER_INPUT)
@lg.publisher(ROLLING_AVERAGER_OUTPUT)
async def average(self, message: RandomMessage) -> lg.AsyncPublisher:
current_time = time.time()
global data
global throughput_data
global datarate_data
global one_second_must_pass

start_time = perf_counter()
current_time_UTC = str(datetime.utcnow()) + " (UTC)"

self.state.messages.append(message)
self.state.messages = [
message
for message in self.state.messages
if message.timestamp >= current_time - self.config.window
if message.timestamp >= time.time() - self.config.window
]
if len(self.state.messages) == 0:
return
all_data = np.stack(
[message.data for message in self.state.messages]
)
mean_data = np.mean(all_data, axis=0)

data.append(message.data.size * message.data.itemsize)
if perf_counter() - one_second_must_pass >= 1:
tempThroughput = 0
tempDataRate = 0
for messagedata in data:
tempThroughput += messagedata
tempDataRate += 1.0
throughput_data = str(tempThroughput) + " bytes"
datarate_data = str(tempDataRate) + " msg/sec"
data.clear()
one_second_must_pass = perf_counter()

latency_data = str(perf_counter() - start_time) + " s"

yield self.ROLLING_AVERAGER_OUTPUT, RandomMessage(
timestamp=current_time,
data=mean_data
timestamp=current_time_UTC,
data=mean_data,

latency=latency_data,
throughput=throughput_data,
datarate=datarate_data
)
34 changes: 26 additions & 8 deletions extensions/prototypes/labgraph_monitor/src/components/SettingPanel/EdgeSettings.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,11 +112,22 @@ const Edge: React.FC = (): JSX.Element => {
(field, index) => {
return (
<TableRow key={index}>
{open && (
<TableCell>
{field[0]}
{open ? (
<TableCell
style={{
whiteSpace:
'normal',
wordBreak:
'break-word',
}}
>
{field[0] ===
'timestamp' ||
field[0] === 'data'
? `${field[0]} `
: null}
</TableCell>
)}
) : null}
{open ? (
<TableCell
style={{
Expand All @@ -127,11 +138,18 @@ const Edge: React.FC = (): JSX.Element => {
}}
>
{connection ===
WS_STATE.CONNECTED
? `${field[1].content}, `
: mockData.join(
WS_STATE.CONNECTED &&
(field[0] ===
'timestamp' ||
field[0] ===
'data')
? `${field[1].content} `
: connection ===
WS_STATE.DISCONNECTED
? mockData.join(
' '
)}
)
: null}
</TableCell>
) : null}
</TableRow>
Expand Down
Loading