This thread has been locked.
If you have a related question, please click the "Ask a related question" button in the top right corner. The newly created question will be automatically linked to this question.
Tool/software:
Hi,
I have flashed the IWR6843ISK-ODS with the prebuilt binary, "3D_people_count_68xx_demo.bin" and the flashing part is fine but when I run the mmWave_People_Counting_GUI.exe I get stuck as to how to actually visualize the data. Am I understanding the flow of the experiment correctly? Or is there another way of visualizing and saving the data?
Thanks,
Erik
Hi,
Is this the experiment you're referencing?
If so, what is the issue with the visualizer? Please send all versions of the binaries and configurations you're using. If you use the wrong toolbox version for the software this visualizer does not work.
Best,
Nate
Hi Nate,
I'm using the 3D_people_count_68xx_demo.bin from the mmwave_industrial_toolbox_4_11_0 and the visualizer is the mmWave_People_Counting_GUI.exe. When I start the visualizer, what should be the UART and DATA COM ports? I only have one cable connected to the IWR67843ISK-ODS eval board.
Thanks,
Erik
Hi Nathan,
I also have a python file that should generate the point cloud data after I wave my hand in front of it, but it seems to be not responding. I've attached the code here
import os
from typing import List
import matplotlib
import numpy as np
from matplotlib.ticker import PercentFormatter
from scipy.spatial.transform import Rotation
from optitrack_visualizer import OptitrackData
matplotlib.use('Qt5Agg') # prevent matplotlib environment error caused by pyside6
import matplotlib.pyplot as plt
plt.ion()
# Settings
FILTER = {
'range': [0, 60], # m
'snr': [0.1, None], # dB
'doppler': [-30, 30], # cm/s
'azimuth': [None, None], # deg
'elevation': [None, None], # deg
}
XYZ_CUBE = 36
PLOT_LIMS = {
'range': [1, 8], # m
'snr': [0.5, 3.5], # dB
'doppler': [-20, 20], # cm/s
'azimuth': [-75, 75], # deg
'elevation': [-75, 75], # deg
'x': [-XYZ_CUBE / 2, XYZ_CUBE / 2], # m (out from radar)
'y': [-XYZ_CUBE / 2, XYZ_CUBE / 2], # m (side to side)
'z': [-XYZ_CUBE / 2, XYZ_CUBE / 2], # m (up and down)
'histogram': [0, 0.15], # *100%
'timestamp': [None, None], # s
'num_points': [0, 110],
'position': [-4, 4],
}
SCALE_FIGURE_SIZE = 1
APPLY_COORDINATE_TRANSFORMATION = True
# RADAR_POSITION = (-3.160669842779445, -0.2900309809625310, 1.0792819198300263) # xyz [m] (9.10)
RADAR_POSITION = (-3.164462045344589, -0.4513652662256102, 0.9339563475978306) # xyz [m] (9.11)
RADAR_ANGLE_UP = 0 # deg
RADAR_ANGLE_RIGHT = 0 # deg
TIME_OFFSET = 0 # seconds
def main():
folder_path = r'C:\Users\EE36381\Documents\Python_Scripts\FOLDER'
viz = DynamicVisualizer()
for filename in os.listdir(folder_path):
if filename.endswith('.csv'):
viz.add_data(os.path.join(folder_path, filename))
viz.plot_y_vs_x('num_points', 'range', 'range', max_filter=None, trendline=False, show=False)
# vicon = OptitrackData(folder='2024.09.10 - CAVE Lab vicon data', name='vicon', filename_contains='Mocksat')
# viz.plot_position_v_time(show=False, vicon_data=vicon)
# viz.plot_position_v_time_error(show=False, vicon_data=vicon)
class DynamicVisualizer:
def __init__(self):
self.figure_initialized = False
self.fig = None
self.ax = None
self.data = {}
self.error_statistics = {}
self.init_fig()
def init_fig(self, subplots: List[int] = None) -> None:
if not self.figure_initialized:
self.fig = plt.figure(figsize=(12 * SCALE_FIGURE_SIZE, 8 * SCALE_FIGURE_SIZE), tight_layout=True)
if subplots is None:
self.ax = self.fig.add_subplot(111)
else:
self.ax = []
for plot in subplots:
self.ax.append(self.fig.add_subplot(plot))
self.figure_initialized = True
def close_fig(self) -> None:
if self.figure_initialized:
plt.close(self.fig)
self.figure_initialized = False
def add_data(self, csv_filename: str) -> None:
frames = Frames()
cloud = Cloud()
t0 = 0
with open(csv_filename, 'r') as f:
header = f.readline() # range_m, elevation_rad, azimuth_rad, doppler_m/s, snr
while True:
line = f.readline()
if not line:
break
if line[0] == '*':
if len(cloud) > 0:
frames.append(cloud.__copy__())
cloud = Cloud()
try:
timestamp, frame_num = line[1:].split(',')
timestamp = float(timestamp)
frame_num = float(frame_num)
except ValueError:
timestamp = 0
frame_num = 0
if t0 is None:
t0 = timestamp
continue
line_vals = line.strip().replace('\n', '').replace('\r', '').split(',')
with np.errstate(divide='raise'):
try:
cloud.append(Point(*[float(x) for x in line_vals], timestamp - t0, frame_num))
except FloatingPointError:
pass
self.data[os.path.basename(csv_filename)] = frames
def plot_y_vs_x(self, y_name: str, x_name: str, c_name: str, max_filter: str = None, trendline: bool = False,
show: bool = False) -> None:
if max_filter is not None:
max_filter_str = f'Highest {max_filter} point per frame'
else:
max_filter_str = 'All points'
for filename, frames in self.data.items():
if max_filter is not None:
filtered_frames = frames.filter(FILTER)
x = frames.weighted_avg(x_name, max_filter)
y = frames.weighted_avg(y_name, max_filter)
c = frames.weighted_avg(c_name, max_filter)
else:
filtered_frames = frames.filter(FILTER)
filtered_points = filtered_frames.get_points()
x = [getattr(p, x_name) for p in filtered_points]
y = [getattr(p, y_name) for p in filtered_points]
c = [getattr(p, c_name) for p in filtered_points]
self.init_fig()
scatter = self.ax.scatter(x, y, c=c, vmin=PLOT_LIMS[c_name][0], vmax=PLOT_LIMS[c_name][1], zorder=1)
cbar = self.fig.colorbar(scatter, ax=self.ax, label=LABELS[c_name])
if trendline:
trendline_func = np.poly1d(np.polyfit(x, y, 1))
self.ax.plot(
x,
trendline_func(x),
color='red',
linestyle='--',
label=f'y = {str(trendline_func).strip()}',
zorder=0,
)
self.ax.legend()
self.ax.set_title(filename.replace('.csv', '') + f' | {max_filter_str}\n' + FILTER_STR)
self.ax.set_xlim(PLOT_LIMS[x_name])
self.ax.set_ylim(PLOT_LIMS[y_name])
self.ax.set_xlabel(LABELS[x_name])
self.ax.set_ylabel(LABELS[y_name])
print(filename, f'| min={min(y):.2f}, max={max(y):.2f}')
fname_modifier = f' PLOTS {y_name} vs {x_name}, color={c_name}, {max_filter_str}'
filepath = os.path.join(
folder_path + fname_modifier, 'xy_' + os.path.basename(filename).replace('.csv', '') + '.png'
)
if not os.path.exists(os.path.dirname(filepath)):
os.makedirs(os.path.dirname(filepath))
self.fig.savefig(filepath)
if show:
plt.pause(0.001) # Allows GUI to update
self.close_fig()
def plot_position_v_time(self, show: bool = False, vicon_data: 'OptitrackData' = None) -> None:
for filename, frames in self.data.items():
filtered_frames = frames.filter(FILTER)
t = filtered_frames.weighted_avg('timestamp', 'snr')
x = filtered_frames.weighted_avg('x', 'snr')
y = filtered_frames.weighted_avg('y', 'snr')
z = filtered_frames.weighted_avg('z', 'snr')
self.init_fig()
self.ax.plot(t, x, '.', label='Radar x', color='tab:blue')
self.ax.plot(t, y, '.', label='Radar y', color='tab:orange')
self.ax.plot(t, z, '.', label='Radar z', color='tab:green')
if vicon_data is not None:
xlims = self.ax.get_xlim()
vicon_data.plot_position_vs_time(fig=self.fig, ax=self.ax, format_plot=False)
self.ax.set_xlim(xlims)
self.ax.legend(loc='upper left')
self.ax.set_xlabel('Time [s]')
self.ax.set_ylabel('Position [m]')
self.ax.set_title(filename + ' | Mean Position over Time')
self.ax.set_ylim(PLOT_LIMS['position'])
print(filename + ' | Mean Position over Time')
filepath = os.path.join(
folder_path + ' PLOTS position vs time', 'position_' + filename.replace('.csv', '') + '.png'
)
if not os.path.exists(os.path.dirname(filepath)):
os.makedirs(os.path.dirname(filepath))
self.fig.savefig(filepath)
if show:
plt.pause(0.001) # Allows GUI to update
self.close_fig()
def plot_position_v_time_error(self, vicon_data: 'OptitrackData', show: bool = False) -> None:
for filename, frames in self.data.items():
filtered_frames = frames.filter(FILTER)
t = filtered_frames.weighted_avg('timestamp', 'snr')
x = filtered_frames.weighted_avg('x', 'snr')
y = filtered_frames.weighted_avg('y', 'snr')
z = filtered_frames.weighted_avg('z', 'snr')
x_plot = []
y_plot = []
z_plot = []
for i in range(len(t)):
vx, vy, vz = vicon_data.get_position_at_timestamp(t[i])
x_plot.append(x[i] - vx)
y_plot.append(y[i] - vy)
z_plot.append(z[i] - vz)
self.close_fig()
self.init_fig(subplots=[311, 312, 313])
self.ax[0].plot(t, x_plot, '.', label='x error', color='tab:blue')
self.ax[1].plot(t, y_plot, '.', label='y error', color='tab:orange')
self.ax[2].plot(t, z_plot, '.', label='z error', color='tab:green')
self.fig.suptitle(filename + ' | Residual Error')
self.ax[1].set_ylabel('Residual Error [m]')
self.ax[2].set_xlabel('Time [s]')
for ax in self.ax:
xlims = ax.get_xlim()
ax.plot((xlims[0], xlims[1]), (0, 0), color='lightgray', linewidth=1, zorder=0)
ax.set_xlim(xlims)
ax.set_ylim([-1, 1])
ax.legend(loc='upper right')
print(filename + ' | Residual Error')
filepath = os.path.join(
folder_path + ' PLOTS residual vs time', 'residual_' + filename.replace('.csv', '') + '.png'
)
if not os.path.exists(os.path.dirname(filepath)):
os.makedirs(os.path.dirname(filepath))
self.fig.savefig(filepath)
if show:
plt.pause(0.001) # Allows GUI to update
self.close_fig()
self.error_statistics[filename] = {'x': {}, 'y': {}, 'z': {}}
for stats_key, data in zip(self.error_statistics[filename].keys(), [x_plot, y_plot, z_plot]):
self.error_statistics[filename][stats_key] = {
'avg': np.average(data),
'min': np.min(data),
'max': np.max(data),
'std': np.std(data),
}
self.print_error_stats()
def print_error_stats(self):
print('\nAll data in meters:')
S = ','
s = f'filename{S}x_error_avg{S}x_error_min{S}x_error_max{S}x_error_std_deviation{S}y_error_avg{S}' \
f'y_error_min{S}y_error_max{S}y_error_std_deviation{S}z_error_avg{S}z_error_min{S}z_error_max{S}' \
f'z_error_std_deviation\n'
for filename, data_dict in self.error_statistics.items():
s += os.path.basename(filename).replace('.csv', '')
for axis, stats_dict in data_dict.items():
s += f"{S}{stats_dict['avg']:.3f}{S}{stats_dict['min']:.3f}{S}{stats_dict['max']:.3f}{S}" \
f"{stats_dict['std']:.3f}"
s += '\n'
print(s)
class Point:
def __init__(self, range: float, el: float, az: float, doppler: float, snr: float, timestamp: float,
frame_num: float):
'''mmWave pointcloud data
:param range: distance to object [m]
:param el: elevation angle [rad]
:param az: azimuth angle [rad]
:param doppler: doppler velocity [m/s]
:param snr: signal to noise ratio
:param timestamp: time since epoch [s]
:param frame_num: frame number
'''
# TODO: temporary negative here - this is the typical radar orientation
az = az
el = -el
# Measured values
self.elevation = np.degrees(el) # deg
self.azimuth = np.degrees(az) # deg
self.range = range # m
self.doppler = 100 * doppler # cm/s
self.raw_snr = snr # mV ???
self.timestamp = timestamp # seconds
self.frame_num = frame_num
# Calculated values
# https://computitos.files.wordpress.com/2008/03/cartesian_spherical_transformation.pdf
self.x = range * np.cos(el) * np.cos(az)
self.y = range * np.cos(el) * np.sin(az)
self.z = range * np.sin(el)
self.snr = np.log10(snr) # dB ???
self.num_points = None # Updated later
# Transform xyz data
if APPLY_COORDINATE_TRANSFORMATION:
# Dear user: For some data analysis the following code likely has to be modified for different locations
# or radar orientations. Good luck!
self.rotate(0, -RADAR_ANGLE_UP, 0) # radar is oriented so azimuth is horizontal
# self.swap_axes()
self.apply_offset(RADAR_POSITION)
self.timestamp += TIME_OFFSET
def __repr__(self):
s = f'Point('
for key, val in self.__dict__.items():
s += f'{key}={val}, '
s = s[:-2] + ')'
return s
def swap_axes(self) -> None:
'''Swap the xyz axes to a new xyz coordinate axes. This function will need to be defined depending on the
testing location and setup.
:return: None
'''
x = self.x
y = self.y
z = self.z
# Caltech GPS
# self.x = -x
# self.y = -z
# self.z = y
# CAVE Lab EL horizontal
self.x = x
self.y = -z
self.z = y
def apply_offset(self, xyz: List[float]) -> None:
'''Apply an offset to the Point's xyz position
:param xyz: (x, y, z)
:return: None
'''
self.x += xyz[0]
self.y += xyz[1]
self.z += xyz[2]
def rotate(self, rx: float, ry: float, rz: float):
'''Rotate along euler angles (minimal angle error because the rotation is actually done with quaternions)
:param roll_x: rotate around x axis (degrees)
:param pitch_y: rotate around y axis (degrees)
:param yaw_z: rotate around z axis (degrees)
:return: None
'''
rotation = Rotation.from_euler('xyz', [rx, ry, rz], degrees=True)
vec = rotation.apply([self.x, self.y, self.z])
self.x = vec[0]
self.y = vec[1]
self.z = vec[2]
class Cloud:
'''This class represents one frame from the radar. Each frame contains a single point cloud.'''
def __init__(self, points: List['Point'] = None):
self.points = [] if points is None else points
def __copy__(self):
return type(self)(self.points)
def __len__(self):
return len(self.points)
def __repr__(self):
s = f'Clouds(num_points={len(self.points)})'
return s
def append(self, point: 'Point') -> None:
'''Append a Point to this cloud
:param point: Point object
:return: None
'''
self.points.append(point)
def list(self, attribute: str) -> list:
'''For each point in this cloud, get the value of the given attribute and append it to a list. So the output
is a list of all values in this cloud of a certain attribute. Like a list of all azimuth angles from all the
points in this cloud.
:param attribute: an attribute from the Point class
:return: list
'''
return [getattr(p, attribute) for p in self.points]
def update_num_points(self) -> None:
'''Update the "num_points" field within all points in the cloud
:return: None
'''
for point in self.points:
point.num_points = len(self.points)
def max_point(self, attribute: str) -> 'Point':
'''Get the point with the highest value of a given attribute in this cloud
:param attribute: an attribute from the Point class
:return: Point object
'''
return max(self.points, key=lambda x: getattr(x, attribute))
def filter(self, attribute, minval, maxval) -> 'Cloud':
'''Filter the data in this point cloud, based on Point attribute. Returns a new Cloud object.
:param attribute: an attribute from the Point class
:param minval: remove points with attribute value below this value
:param maxval: remove points with attribute value above this value
:return: Cloud object
'''
if minval is None and maxval is not None:
filter_func = lambda point, attr: (getattr(point, attr) < maxval)
elif minval is not None and maxval is None:
filter_func = lambda point, attr: (getattr(point, attr) > minval)
elif minval is None and maxval is None:
filter_func = lambda point, attr: True
else:
if minval >= maxval:
print(f'ERROR: minval ({minval}) is greater than maxval ({maxval})!')
exit(1)
filter_func = lambda point, attr: (getattr(point, attr) > minval and getattr(point, attr) < maxval)
return Cloud([p for p in self.points if filter_func(p, attribute)])
def weighted_avg(self, avg_attr: str, weight_attr: str) -> float:
'''Computes the weighted average of the attribute across all points in this cloud. Common use case is to get
the average position in x, y, or z, weighted by SNR.
:param avg_attr: attribute to average over
:param weight_attr: attribute to use as the weights
:return: value
'''
data = [getattr(p, avg_attr) for p in self.points]
weights = np.asarray([getattr(p, weight_attr) for p in self.points])
return np.average(data, weights=weights)
class Frames:
'''This class holds all the data returned from the radar for a given test run. In other words, all the data from a
single CSV gets read into here.'''
def __init__(self, clouds: List['Cloud'] = None, t0: float = None):
self.clouds = [] if clouds is None else clouds
self.t0 = 0 if t0 is None else t0
def __repr__(self):
s = f'Frames(num_clouds={len(self.clouds)})'
return s
def append(self, cloud: 'Cloud'):
'''Append a new cloud to this collection of frames
:param cloud: Cloud object
:return: None
'''
cloud.update_num_points()
self.clouds.append(cloud)
def get_points(self):
'''Get list of all Points in this object
:return: list of Point objects
'''
return [point for cloud in self.clouds for point in cloud.points]
def weighted_avg(self, avg_attr: str, weight_attr: str):
'''Get a list of values, where each value is the weighted average of a cloud. Averaging is done according to
the provided attributes.
:param avg_attr: Average over this attribute (from the Point class)
:param weight_attr: Use this attribute as the weights.
:return:
'''
return [c.weighted_avg(avg_attr, weight_attr) for c in self.clouds]
def max_filter(self, max_attr: str, cloud_filter: dict):
'''Get a list of Points, from all data contiained in this object, where each Point is the point with the
highest value of the provided attribute in it's cloud.
cloud_filter = {
'attribute1': (min, max),
'attribute2': (min, max),
...
}
:param max_attr: find the point with the highest value in this attribute
:param cloud_filter: dict, detailed above
:return: list of Point objects
'''
filtered_frames = self.filter(cloud_filter)
return [max(cloud.points, key=lambda x: getattr(x, max_attr)) for cloud in filtered_frames.clouds]
def filter(self, cloud_filter: dict):
'''Returns a new Frames object with all contained points satisfying the provided filter.
cloud_filter = {
'attribute1': (min, max),
'attribute2': (min, max),
...
}
:param cloud_filter: dict, detailed above
:return: new Frames object
'''
out = Frames()
for cloud in self.clouds:
filtered_cloud = cloud
for attr, min_max in cloud_filter.items():
filtered_cloud = filtered_cloud.filter(attr, min_max[0], min_max[1])
if len(filtered_cloud) > 0:
out.append(filtered_cloud)
return out
def count(self) -> int:
'''Get number of clouds in this Frames object
:return: number of clouds
'''
return len(self.clouds)
def get_frame_at_time(self, timestamp, tolerance=1, frames_before=0, frames_after=0):
'''Get one or more frames nearest to a given timestamp from this object. By default, this returns one frame,
but you can specify if you'd like n frames before or after the nearest frame.
:param timestamp: timestamp, in seconds, that you want to get a frame nearest
:param tolerance: the most number of seconds delta between the nearest frame and the requested timestamp
:param frames_before: Also return this many frames before the nearest frame
:param frames_after: Also return this many frames after the nearest frame
:return: new Frames object
'''
# Find the nearest frame that satisfies the tolerance
time_diff = tolerance
nearest_cloud_index = None
for i in range(0, len(self.clouds)):
if abs(self.clouds[i].points[0].timestamp - timestamp) < time_diff:
time_diff = abs(self.clouds[i].points[0].timestamp - timestamp)
nearest_cloud_index = i
# Return if no suitable frame is found
if nearest_cloud_index is None:
return None
# New frames object for the return
frames = Frames()
# Append n frames before
for i in range(1, frames_before + 1):
if nearest_cloud_index - i > 0:
frames.append(self.clouds[nearest_cloud_index - i])
# Append the frame we found
frames.append(self.clouds[nearest_cloud_index])
# Append n frames after
for i in range(1, frames_after + 1):
if nearest_cloud_index + i < len(self.clouds):
frames.append(self.clouds[nearest_cloud_index + i])
return frames
# These are used as axis labels for when the attribute is plotted
LABELS = {
'range': 'Range [m]',
'snr': 'log10(SNR) [dB]',
'doppler': 'Doppler [cm/s]',
'timestamp': 'Time [s]',
'x': 'x [m]',
'y': 'y [m]',
'z': 'z [m]',
'azimuth': 'Azimuth [deg]',
'elevation': 'Elevation [deg]',
'num_points': 'Points per Frame',
}
# Generate a string describing the filter
if FILTER != {}:
FILTER_STR = 'Filter: '
for key, value in FILTER.items():
if value[0] is None and value[1] is None:
# No filter is being applied if both bounds are none, so don't put it in the string
pass
FILTER_STR += f'{key}={value}, '
FILTER_STR = FILTER_STR[:-2]
else:
FILTER_STR = 'Filter: None'
# Main
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
exit(0)
Thanks,
Erik
Hi Erik,
Please use the most updated toolbox found here and the industrial visualizer from the same package for a python reference.
Best,
Nate
Hi Nathan,
Thank you for this, I got the 3D visualizer to work and saw it in realtime. I have a couple of follow-up questions:
Thanks,
Erik
Hi,
The visualizer saves to json by default. You may parse the .json and convert to CSV if you like, or you may modify the python visualizer to output a CSV file. It is written in fully open source python for you to modify as your please.
Best
Nate
Hi Nathan,
I got the visualizer from the radar toolbox to output the json files and I wrote a script to convert it to CSV, I just want to make sure Im understanding the pointcloud[] data correctly. I've attached a sample from one of my own json outputs:
"demo": "3D People Tracking",
"device": "xWR6843",
"data": [
{
"frameData": {
"error": 0,
"frameNum": 1,
"pointCloud": []
},
"timestamp": 1751307022340.4265
},
{
"frameData": {
"error": 0,
"frameNum": 2,
"pointCloud": [
[
-0.3930234318389437,
0.2635885666207569,
1.8032516212942036,
-2.8347200757125393,
5.119999885559082,
0.0,
255.0
]
],
"numDetectedPoints": 1
Thanks,
Erik
Hi,
that should be x,y,z,velocity,snr,noise,trackID. Coordinates are local to the radar. See the page here
Best
Nate