Hello Ti,
we have written python code to check fall detection in IWR6843AOPEVM radar module.
Code is running but only for once after that it get crashed.
Code is attached here
# Standard Imports
import sys
import numpy as np
import time
import math
import struct
import os
import string
import serial
import serial.tools.list_ports
import statistics
import warnings
import random
import copy
import collections
from collections import deque
from gl_classes import GLTextItem
# Local File Imports
from gui_threads import *
from gui_parser import uartParser
from graphUtilities import *
from gui_common import *
from cachedData import *
from fall_detection import FallDetection, fallDetectionSliderClass
# from PyQt5.QtWidgets import (QApplication, QCheckBox, QComboBox, QDateTimeEdit,
# QDial, QDialog, QGridLayout, QGroupBox, QHBoxLayout, QLabel, QLineEdit,
# QProgressBar, QPushButton, QRadioButton, QScrollBar, QSizePolicy,
# QSlider, QSpinBox, QStyleFactory, QTableWidget, QTableWidgetItem, QTabWidget, QTextEdit,
# QVBoxLayout, QWidget, QFileDialog, QButtonGroup, QFormLayout, QFrame, QSpacerItem)
cachedData = cachedDataType()
class Window():
def __init__(self, parent=None, size=[]):
# TODO bypass serial read function to also log to a file
self.frameTime = 50
self.graphFin = 1
self.hGraphFin = 1
self.threeD = 1
self.lastFramePoints = np.zeros((5,1))
self.plotTargets = 1
self.frameNum = 0
self.profile = {'startFreq': 60.25, 'numLoops': 64, 'numTx': 3, 'sensorHeight':2.0, 'maxRange':10, 'az_tilt':0, 'elev_tilt':15.0, 'enabled':0}
self.chirpComnCfg = {'DigOutputSampRate':23, 'DigOutputBitsSel':0, 'DfeFirSel':0, 'NumOfAdcSamples':128, 'ChirpTxMimoPatSel':4, 'ChirpRampEndTime':36.1, 'ChirpRxHpfSel':1}
self.chirpTimingCfg = {'ChirpIdleTime':8, 'ChirpAdcSkipSamples':24, 'ChirpTxStartTime':0, 'ChirpRfFreqSlope':47.95, 'ChirpRfFreqStart':60}
self.guiMonitor = {'pointCloud':1, 'rangeProfile':0, 'NoiseProfile':0, 'rangeAzimuthHeatMap':0, 'rangeDopplerHeatMap':0, 'statsInfo':0}
self.rangeRes = 0
self.rangeAxisVals = np.zeros(int(self.chirpComnCfg['NumOfAdcSamples']/2))
self.sensorHeight = 1.5
self.numFrameAvg = 10
self.configSent = 0
self.previousFirstZ = -1
self.yzFlip = 0
self.trackColorMap = None
self.prevConfig = DEMO_NAME_OOB
self.vitalsPatientData = []
self.pointBounds = {'enabled':False, 'minX':0, 'maxX':0, 'minY':0, 'maxY':0, 'minZ':0, 'maxZ':0}
# Flag to indicate if the last frame was parsed incorrectly to ignore the error on the subsequent frame when
# the number of points won't be consistent with the tracker data
self.lastFrameErrorFlag = False
#color gradients
# # TODO Simplify color gradients
# self.Gradients = OrderedDict([
# ('bw', {'ticks': [(0.0, (0, 0, 0, 255)), (1, (255, 255, 255, 255))], 'mode': 'rgb'}),
# ('hot', {'ticks': [(0.3333, (185, 0, 0, 255)), (0.6666, (255, 220, 0, 255)), (1, (255, 255, 255, 255)), (0, (0, 0, 0, 255))], 'mode': 'rgb'}),
# ('jet', {'ticks': [(1, (166, 0, 0, 255)), (0.32247191011235954, (0, 255, 255, 255)), (0.11348314606741573, (0, 68, 255, 255)), (0.6797752808988764, (255, 255, 0, 255)), (0.902247191011236, (255, 0, 0, 255)), (0.0, (0, 0, 166, 255)), (0.5022471910112359, (0, 255, 0, 255))], 'mode': 'rgb'}),
# ('summer', {'ticks': [(1, (255, 255, 0, 255)), (0.0, (0, 170, 127, 255))], 'mode': 'rgb'} ),
# ('space', {'ticks': [(0.562, (75, 215, 227, 255)), (0.087, (255, 170, 0, 254)), (0.332, (0, 255, 0, 255)), (0.77, (85, 0, 255, 255)), (0.0, (255, 0, 0, 255)), (1.0, (255, 0, 127, 255))], 'mode': 'rgb'}),
# ('winter', {'ticks': [(1, (0, 255, 127, 255)), (0.0, (0, 0, 255, 255))], 'mode': 'rgb'}),
# ('spectrum2', {'ticks': [(1.0, (255, 0, 0, 255)), (0.0, (255, 0, 255, 255))], 'mode': 'hsv'}),
# ('heatmap', {'ticks': [ (1, (255, 0, 0, 255)), (0, (131, 238, 255, 255))], 'mode': 'hsv'})
# ])
# cmap = 'heatmap'
# if (cmap in self.Gradients):
# self.gradientMode = self.Gradients[cmap]
# self.zRange = [-3, 3]
# self.plotHeights = 1
# # Gui size
# if (size):
# left = 50
# top = 50
# width = math.ceil(size.width()*0.9)
# height = math.ceil(size.height()*0.9)
# self.setGeometry(left, top, width, height)
# # Persistent point cloud
# self.previousClouds = []
# self.hearPlotData = []
# self.breathPlotData = []
# # Set up graph pyqtgraph
# self.init3dGraph()
# self.initColorGradient()
# self.init1dGraph()
# # Add connect options
# self.initConnectionPane()
# self.initStatsPane()
# self.initPlotControlPane()
# self.initFallDetectPane()
# self.initConfigPane()
# self.initSensorPositionPane()
# self.initBoundaryBoxPane()
# self.initVitalsPlots()
# # Set the layout
# # Create tab for different graphing options
# self.graphTabs = QTabWidget()
# self.graphTabs.addTab(self.pcplot, '3D Plot')
# self.graphTabs.addTab(self.rangePlot, 'Range Plot')
# self.graphTabs.currentChanged.connect(self.whoVisible)
# self.gridlay = QGridLayout()
# self.gridlay.addWidget(self.comBox, 0,0,1,1)
# self.gridlay.addWidget(self.statBox, 1,0,1,1)
# self.gridlay.addWidget(self.configBox,2,0,1,1)
# self.gridlay.addWidget(self.plotControlBox,3,0,1,1)
# self.gridlay.addWidget(self.fallDetectionOptionsBox,4,0,1,1)
# self.gridlay.addWidget(self.spBox,5,0,1,1)
# self.gridlay.addWidget(self.boxTab,6,0,1,1)
# self.gridlay.setRowStretch(7,1) # Added to preserve spacing
# self.gridlay.addWidget(self.graphTabs,0,1,8,1)
# self.gridlay.addWidget(self.colorGradient, 0, 2, 8, 1)
# self.gridlay.addWidget(self.vitalsPane, 0, 3, 8, 1)
# self.vitalsPane.setVisible(False)
# self.gridlay.setColumnStretch(0,1)
# self.gridlay.setColumnStretch(1,3)
# self.setLayout(self.gridlay)
# Set up parser
self.parser = uartParser(type="3D People Tracking")
# # Check cached data for previously used demo and device to set as default options
# deviceName = cachedData.getCachedDeviceName()
# if (deviceName != ""):
# try:
# self.deviceType.setCurrentIndex(DEVICE_LIST.index(deviceName))
# if (deviceName == "IWR6843" or deviceName == "IWR1843"):
# self.parserType = "DoubleCOMPort"
# elif (deviceName == "IWRL6432"):
# self.parserType = "SingleCOMPort"
# elif (deviceName == "IWRL1432"):
# self.parserType = "SingleCOMPort"
# except:
# print("Device not found. Using default option")
# self.deviceType.setCurrentIndex(0)
# demoName = cachedData.getCachedDemoName()
# if (demoName != ""):
# try:
# if (self.deviceType.currentText() in DEVICE_LIST[0:2]):
# self.configType.setCurrentIndex(x843_DEMO_TYPES.index(demoName))
# if (self.deviceType.currentText() in DEVICE_LIST[2:3]):
# self.configType.setCurrentIndex(x432_DEMO_TYPES.index(demoName))
# if (self.deviceType.currentText() in DEVICE_LIST[3:]):
# self.configType.setCurrentIndex(xWRL1432_DEMO_TYPES.index(demoName))
# except:
# print("Demo not found. Using default option")
# self.configType.setCurrentIndex(0)
def initConnectionPane(self):
# Find all Com Ports
serialPorts = list(serial.tools.list_ports.comports())
# Find default CLI Port and Data Port
for port in serialPorts:
print(port.description)
if (CLI_XDS_SERIAL_PORT_NAME in port.description or CLI_SIL_SERIAL_PORT_NAME in port.description):
print(f'\tCLI COM Port found: {port.device}')
comText = port.device
comText = comText.replace("COM", "")
elif (DATA_XDS_SERIAL_PORT_NAME in port.description or DATA_SIL_SERIAL_PORT_NAME in port.description):
print(f'\tData COM Port found: {port.device}')
comText = port.device
comText = comText.replace("COM", "")
def startApp(self):
self.configSent = 1
self.parseTimer.start(int(self.frameTime)) # need this line
def parseData(self):
self.uart_thread.start(priority=QThread.HighestPriority)
def connectCom(self):
self.parser.frameTime = self.frameTime
# init threads and timers
self.uart_thread = parseUartThread(self.parser)
self.uart_thread.fin.connect(self.parseData)
self.parseTimer = QTimer()
self.parseTimer.setSingleShot(False)
self.parseTimer.timeout.connect(self.parseData)
uart = "COM8"
data = "COM7"
self.parser.connectComPorts(uart, data)
# Gets called whenever the sensor position box is modified
def onChangeSensorPosition(self):
newHeight = float(2.0)
newAzTilt = float(0.0)
newElevTilt = float(15.0)
# self.cThread = sendCommandThread(self.parser,command)
# self.cThread.start(priority=QThread.HighestPriority-2)
# Update Profile info
self.profile['sensorHeight'] = newHeight
# # Move evmBox to new position
# self.evmBox.resetTransform()
# self.evmBox.rotate(-1*newElevTilt,1,0,0)
# self.evmBox.rotate(-1*newAzTilt,0,0,1)
# self.evmBox.translate(0,0,newHeight)
def parseCfg(self, fname):
with open(fname, 'r') as cfg_file:
self.cfg = cfg_file.readlines()
counter = 0
chirpCount = 0
for line in self.cfg:
args = line.split()
if (len(args) > 0):
# cfarCfg
if (args[0] == 'cfarCfg'):
pass
#self.cfarConfig = {args[10], args[11], '1'}
# trackingCfg
elif (args[0] == 'trackingCfg'):
if (len(args) < 5):
print ("Error: trackingCfg had fewer arguments than expected")
continue
self.profile['maxTracks'] = int(args[4])
# Update the maximum number of tracks based off the cfg file
# self.trackColorMap = get_trackColors(self.profile['maxTracks'])
for m in range(self.profile['maxTracks']):
# Add track gui object
mesh = gl.GLLinePlotItem()
mesh.setVisible(False)
# self.pcplot.addItem(mesh)
# self.ellipsoids.append(mesh)
# Add track coordinate string
# text = GLTextItem()
# text.setGLViewWidget(self.pcplot)
# text.setVisible(False)
# self.pcplot.addItem(text)
# self.coordStr.append(text)
# Add track classifier label string
# classifierText = GLTextItem()
# classifierText.setGLViewWidget(self.pcplot)
# classifierText.setVisible(False)
# self.pcplot.addItem(classifierText)
# self.classifierStr.append(classifierText)
# If we only support 1 patient, hide the other patient window
# if (self.profile['maxTracks'] == 1):
# self.vitals[1]['pane'].setVisible(False)
# Initialize Vitals output dictionaries for each potential patient
# for i in range (min(self.profile['maxTracks'], MAX_VITALS_PATIENTS)):
# # Initialize
# patientDict = {}
# patientDict ['id'] = i
# patientDict ['rangeBin'] = 0
# patientDict ['breathDeviation'] = 0
# patientDict ['heartRate'] = []
# patientDict ['breathRate'] = 0
# patientDict ['heartWaveform'] = []
# patientDict ['breathWaveform'] = []
# self.vitalsPatientData.append(patientDict)
# # Make each patient's pane visible
# self.vitals[i]['pane'].setVisible(True)
elif (args[0] == 'AllocationParam'):
pass
#self.allocConfig = tuple(args[1:6])
elif (args[0] == 'GatingParam'):
pass
#self.gatingConfig = tuple(args[1:4])
elif (args[0] == 'SceneryParam' or args[0] == 'boundaryBox'):
if (len(args) < 7):
print ("Error: SceneryParam/boundaryBox had fewer arguments than expected")
continue
self.boundaryLine = counter
leftX = float(args[1])
rightX = float(args[2])
nearY = float(args[3])
farY = float(args[4])
bottomZ = float(args[5])
topZ = float(args[6])
# self.addBoundBox('trackerBounds', leftX, rightX, nearY, farY, bottomZ, topZ)
# Default pointBounds box to have the same values as the last boundaryBox in the config
# These can be changed by the user
# self.boundaryBoxes[0]['boundList'][0].setText(str(leftX))
# self.boundaryBoxes[0]['boundList'][1].setText(str(rightX))
# self.boundaryBoxes[0]['boundList'][2].setText(str(nearY))
# self.boundaryBoxes[0]['boundList'][3].setText(str(farY))
# self.boundaryBoxes[0]['boundList'][4].setText(str(bottomZ))
# self.boundaryBoxes[0]['boundList'][5].setText(str(topZ))
elif (args[0] == 'staticBoundaryBox'):
self.staticLine = counter
elif (args[0] == 'profileCfg'):
if (len(args) < 12):
print ("Error: profileCfg had fewer arguments than expected")
continue
self.profile['startFreq'] = float(args[2])
self.profile['idle'] = float(args[3])
self.profile['adcStart'] = float(args[4])
self.profile['rampEnd'] = float(args[5])
self.profile['slope'] = float(args[8])
self.profile['samples'] = float(args[10])
self.profile['sampleRate'] = float(args[11])
print(self.profile)
elif (args[0] == 'frameCfg'):
if (len(args) < 4):
print ("Error: frameCfg had fewer arguments than expected")
continue
self.frameTime = float(args[5])
self.profile['numLoops'] = float(args[3])
self.profile['numTx'] = float(args[2])+1
elif (args[0] == 'chirpCfg'):
chirpCount += 1
elif (args[0] == 'sensorPosition'):
# sensorPosition for x843 family has 3 args
self.profile['sensorHeight'] = float(args[1])
self.profile['az_tilt'] = float(args[2])
self.profile['elev_tilt'] = float(args[3])
print("Juber")
# # sensorPosition for x432 family has 5 args
# if (self.deviceType.currentText() in DEVICE_LIST[2] or self.deviceType.currentText() in DEVICE_LIST[3]):
# if (len(args) < 6):
# print ("Error: sensorPosition had fewer arguments than expected")
# continue
# #xOffset and yOffset are not implemented in the python code yet.
# self.profile['xOffset'] = float(args[1])
# self.profile['yOffset'] = float(args[2])
# self.profile['sensorHeight'] = float(args[3])
# self.profile['az_tilt'] = float(args[4])
# self.profile['elev_tilt'] = float(args[5])
# Only used for Small Obstacle Detection
elif (args[0] == 'occStateMach'):
numZones = int(args[1])
if (numZones > 2):
print('ERROR: More zones specified by cfg than are supported in this GUI')
# Only used for Small Obstacle Detection
elif (args[0] == 'zoneDef'):
if (len(args) < 8):
print ("Error: zoneDef had fewer arguments than expected")
continue
zoneIdx = int(args[1])
minX = float(args[2])
maxX = float(args[3])
minY = float(args[4])
maxY = float(args[5])
# Offset by 3 so it is in center of screen
minZ = float(args[6]) + self.profile['sensorHeight']
maxZ = float(args[7]) + self.profile['sensorHeight']
name = 'occZone' + str(zoneIdx)
# self.addBoundBox(name, minX, maxX, minY, maxY, minZ, maxZ)
elif (args[0] == 'mpdBoundaryBox'):
if (len(args) < 8):
print ("Error: mpdBoundaryBox had fewer arguments than expected")
continue
zoneIdx = int(args[1])
minX = float(args[2])
maxX = float(args[3])
minY = float(args[4])
maxY = float(args[5])
minZ = float(args[6])
maxZ = float(args[7])
name = 'mpdBox' + str(zoneIdx)
# self.addBoundBox(name, minX, maxX, minY, maxY, minZ, maxZ)
elif (args[0] == 'chirpComnCfg'):
if (len(args) < 8):
print ("Error: chirpComnCfg had fewer arguments than expected")
continue
try:
self.chirpComnCfg['DigOutputSampRate'] = int(args[1])
self.chirpComnCfg['DigOutputBitsSel'] = int(args[2])
self.chirpComnCfg['DfeFirSel'] = int(args[3])
self.chirpComnCfg['NumOfAdcSamples'] = int(args[4])
self.chirpComnCfg['ChirpTxMimoPatSel'] = int(args[5])
self.chirpComnCfg['ChirpRampEndTime'] = 10 * float(args[6])
self.chirpComnCfg['ChirpRxHpfSel'] = int(args[7])
except Exception as e:
print (e)
elif (args[0] == 'chirpTimingCfg'):
if (len(args) < 6):
print ("Error: chirpTimingCfg had fewer arguments than expected")
continue
self.chirpTimingCfg['ChirpIdleTime'] = 10.0 * float(args[1])
self.chirpTimingCfg['ChirpAdcSkipSamples'] = int(args[2]) << 10
self.chirpTimingCfg['ChirpTxStartTime'] = 10.0 * float(args[3])
self.chirpTimingCfg['ChirpRfFreqSlope'] = float(args[4])
self.chirpTimingCfg['ChirpRfFreqStart'] = float(args[5])
elif (args[0] == 'clusterCfg'):
if (len(args) < 4):
print ("Error: clusterCfg had fewer arguments than expected")
continue
self.profile['enabled'] = float(args[1])
self.profile['maxDistance'] = float(args[2])
self.profile['minPoints'] = float(args[3])
# This is specifically guiMonitor for 60Lo, this parsing will break the gui when an SDK 3 config is sent
elif (args[0] == 'guiMonitor'):
if(self.deviceType.currentText() in DEVICE_LIST[2] or self.deviceType.currentText() in DEVICE_LIST[3] ):
if (len(args) < 12):
print ("Error: guiMonitor had fewer arguments than expected")
continue
self.guiMonitor['pointCloud'] = int(args[1])
self.guiMonitor['rangeProfile'] = int(args[2])
self.guiMonitor['NoiseProfile'] = int(args[3])
self.guiMonitor['rangeAzimuthHeatMap'] = int(args[4])
self.guiMonitor['rangeDopplerHeatMap'] = int(args[5])
self.guiMonitor['statsInfo'] = int(args[6])
counter += 1
# self.rangeRes = (3e8*(100/self.chirpComnCfg['DigOutputSampRate']))/(2*self.chirpTimingCfg['ChirpRfFreqSlope']*self.chirpComnCfg['NumOfAdcSamples'])
self.rangeRes = (3e8*(100/self.chirpComnCfg['DigOutputSampRate'])*1e6)/(2*self.chirpTimingCfg['ChirpRfFreqSlope']*1e12*self.chirpComnCfg['NumOfAdcSamples'])
# self.rangePlot.setXRange(0,(self.chirpComnCfg['NumOfAdcSamples']/2)*self.rangeRes,padding=0.01)
self.rangeAxisVals = np.arange(0, self.chirpComnCfg['NumOfAdcSamples']/2*self.rangeRes, self.rangeRes)
print(self.guiMonitor['rangeProfile'])
# if (self.guiMonitor['rangeProfile'] == 0):
# self.rangePlot.getPlotItem().setLabel('top','range profile disabled')
# elif (self.guiMonitor['rangeProfile'] == 1):
# self.rangePlot.getPlotItem().setLabel('top','Major Range Profile')
# elif (self.guiMonitor['rangeProfile'] == 2):
# self.rangePlot.getPlotItem().setLabel('top','Minor Range Profile')
# elif (self.guiMonitor['rangeProfile'] == 3):
# self.rangePlot.getPlotItem().setLabel('top','Major & Minor Range Profile Mode Not Supported')
# else:
# self.rangePlot.getPlotItem().setLabel('top','INVALID gui monitor range profile input')
# self.profile['maxRange'] = self.profile['sampleRate']*1e3*0.9*3e8/(2*self.profile['slope']*1e12)
# bw = self.profile['samples']/(self.profile['sampleRate']*1e3)*self.profile['slope']*1e12
# rangeRes = 3e8/(2*bw)
# Tc = (self.profile['idle']*1e-6 + self.profile['rampEnd']*1e-6)*chirpCount
# lda = 3e8/(self.profile['startFreq']*1e9)
# maxVelocity = lda/(4*Tc)
# velocityRes = lda/(2*Tc*self.profile['numLoops']*self.profile['numTx'])
# self.configTable.setItem(1,1,QTableWidgetItem(str(self.profile['maxRange'])[:5]))
# self.configTable.setItem(2,1,QTableWidgetItem(str(rangeRes)[:5]))
# self.configTable.setItem(3,1,QTableWidgetItem(str(maxVelocity)[:5]))
# self.configTable.setItem(4,1,QTableWidgetItem(str(velocityRes)[:5]))
# Update sensor position
# self.az_tilt.setText(str(0.0))
# self.elev_tilt.setText(str(15.0))
# self.s_height.setText(str(2.0))
self.onChangeSensorPosition()
def parseData(self):
self.uart_thread.start(priority=QThread.HighestPriority)
def selectCfg(self):
try:
file = "C:/ti/radar_toolbox_1_20_00_11/tools/visualizers/Industrial_Visualizer/AOP_6m_default.cfg"
self.parseCfg(file)
print("Config Loaded")
# if 'maxTracks' in self.profile:
# print('maxTracks')
# self.updateNumTracksBuffer() # Update the max number of tracks based off the config file
except Exception as e:
print(e)
print('No cfg file selected!')
def sendCfg(self):
try:
self.parser.sendCfg(self.cfg)
self.configSent = 1
print("frameTime", self.frameTime)
self.parseTimer.start(int(self.frameTime)) # need this line
except Exception as e:
print(e)
print ('No cfg file selected!')
if __name__ == '__main__':
radar= Window()
radar.initConnectionPane()
radar.connectCom()
radar.selectCfg()
radar.sendCfg()
radar.startApp()
