Part Number: BOOSTXL-AOA
Hi,
I have changed the mode as AOA_MODE_RAW in python file, attached below still i am getting the angle data.
I have made the changes in npi task.c file as mentioned in Task 4
To achieve this, there are two options:
-
Increase the connection interval. In the
rtls_masterreadme file, it's mentioned that the connection interval should be larger than 300ms to acommodate outputting all the samples. -
Increase the UART baurate to 2x, 4x or 8x 115200.
To achieve this, navigate to source/ti/blestack/npi/src/unified/npi_task.c::uint8_t NPITask_Params_init(uint8_t portType, NPI_Params *params) and add the following under the
#if defined(NPI_USE_UART)After making these changes, still i am getting angle as results
Please suggest me how to obtain those
Code:
import queue
import csv
import time
from collections import namedtuple
from rtls import RTLSManager, RTLSNode
# Un-comment the below to get raw serial transaction logs
# import logging, sys
# logging.basicConfig(stream=sys.stdout, level=logging.DEBUG,
# format='[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s')
if __name__ == '__main__':
# Initialize, but don't start RTLS Nodes to give to the RTLSManager
my_nodes = [RTLSNode('COM17', 115200), RTLSNode('COM9', 115200)]
# Prepare csv file to save data
filename = 'rtls_raw_iq_samples.csv'
outfile = open(filename, 'w', newline='')
csv_fieldnames = ['pkt', 'sample_idx', 'rssi', 'ant_array', 'channel', 'i', 'q']
SampleRow = namedtuple('CsvRow', csv_fieldnames)
csv_writer = csv.DictWriter(outfile, fieldnames=csv_fieldnames)
csv_writer.writeheader()
# Temporary storage of iq samples
dump_rows = []
# How many AoA sample buffers should be stored.
# None means infinite. Press Ctrl+C to terminate in this case.
pkt_limit = 200 # None # 5
# Running packet counter
pkt_cnt = 0
# Initialize references to the connected devices
master_node = None
passive_nodes = []
# Initialize references to the connected devices
address = None
address_type = None
# ToF related settings
samples_per_burst = 256 # Should be a power of 2. Hint: in a 100ms interval, there are about 300~ samples
tof_freq_list = [2408, 2412, 2418, 2424] #Other options: 2414, 2420
tof_num_freq = len(tof_freq_list)
auto_tof_rssi = -55
tof_sample_mode = 'TOF_MODE_DIST'
tof_run_mode = 'TOF_MODE_CONT'
seed = 0
samplesPerFreq = 1000
calibDistance = 1 # 1 meter
# AoA related settings
aoa_run_mode = 'AOA_MODE_RAW'
aoa_cte_scan_ovs = 4
aoa_cte_offset = 4
aoa_cte_time = 20
# Auto detect AoA or ToF support related
tof_supported = False
aoa_supported = False
# If slave addr is None, the script will connect to the first RTLS slave
# that it found. If you wish to connect to a specific device
# (in the case of multiple RTLS slaves) then you may specify the address
# explicitly as given in the comment to the right
slave_addr = None #'54:6C:0E:A0:47:43'
# Initialize manager reference, because on Exception we need to stop the manager to stop all the threads.
manager = None
try:
# Start an RTLSManager instance without WebSocket server enabled
manager = RTLSManager(my_nodes, websocket_port=None)
# Create a subscriber object for RTLSManager messages
subscriber = manager.create_subscriber()
# Tell the manager to automatically distribute connection parameters
manager.auto_params = True
# Start RTLS Node threads, Serial threads, and manager thread
manager.start()
# Wait until nodes have responded to automatic identify command and get reference
# to single master RTLSNode and list of passive RTLSNode instances
master_node, passive_nodes, failed = manager.wait_identified()
if len(failed):
print(f"ERROR: {len(failed)} nodes could not be identified. Are they programmed?")
# Exit if no master node exists
if not master_node:
raise RuntimeError("No RTLS Master node connected")
# Combined list for lookup
all_nodes = passive_nodes + [master_node]
# Initialize application variables on nodes
for node in all_nodes:
node.tof_initialized = False
node.seed_initialized = False
node.aoa_initialized =True
#
# At this point the connected devices are initialized and ready
#
# Display list of connected devices and their capabilities
print(f"{master_node.identifier} {', '.join([cap for cap, available in master_node.capabilities.items() if available])}")
# Iterate over Passives and detect their capabilities
for pn in passive_nodes:
print(f"{pn.identifier} {', '.join([cap for cap, available in pn.capabilities.items() if available])}")
# Check over aggregated capabilities to see if they make sense
capabilities_per_node = [[cap for cap, avail in node.capabilities.items() if avail] for node in all_nodes]
tof_supported = all('TOF_PASSIVE' in node_caps or 'TOF_MASTER' in node_caps for node_caps in capabilities_per_node)
# Assume AoA if all nodes are not ToF
aoa_supported = all(not ('TOF_PASSIVE' in node_caps or 'TOF_MASTER' in node_caps) for node_caps in capabilities_per_node)
# Check that Nodes all must be either AoA or ToF
if not (tof_supported or aoa_supported):
raise RuntimeError("All nodes must be either AoA or ToF")
# Need at least 1 passive for AoA
if aoa_supported and len(passive_nodes) == 0:
raise RuntimeError('Need at least 1 passive for AoA')
# Send an example command to each of them, from commands listed at the bottom of rtls/ss_rtls.py
for n in all_nodes:
n.rtls.identify()
while True:
# Get messages from manager
try:
identifier, msg_pri, msg = subscriber.pend(block=True, timeout=0.05).as_tuple()
# Get reference to RTLSNode based on identifier in message
sending_node = manager[identifier]
if sending_node in passive_nodes:
print(f"PASSIVE: {identifier} --> {msg.as_json()}")
else:
print(f"MASTER: {identifier} --> {msg.as_json()}")
# If we received an assert, print it.
if msg.command == 'UTIL_NPI_HW_ASSERT' and msg.type == 'AsyncReq':
raise RuntimeError(f"Received HCI H/W Assert with code: {msg.payload.subcause}")
# After identify is received, we start scanning
if msg.command == 'RTLS_CMD_IDENTIFY':
master_node.rtls.scan()
# Once we start scaning, we will save the address of the
# last scan response
if msg.command == 'RTLS_CMD_SCAN' and msg.type == 'AsyncReq':
address = msg.payload.addr
address_type = msg.payload.addrType
# Once the scan has stopped and we have a valid address, then
# connect
if msg.command == 'RTLS_CMD_SCAN_STOP':
if address is not None and address_type is not None and (slave_addr is None or slave_addr == address):
master_node.rtls.connect(address_type, address)
else:
# If we didn't find the device, keep scanning.
master_node.rtls.scan()
# Once we are connected, then we can do stuff
if msg.command == 'RTLS_CMD_CONNECT' and msg.type == 'AsyncReq':
if msg.payload.status == 'RTLS_SUCCESS':
if tof_supported:
# Find the role based on capabilities of sending node
role = 'TOF_MASTER' if sending_node.capabilities.get('TOF_MASTER', False) else 'TOF_PASSIVE'
# Send the ToF parameters to the node that just connected
sending_node.rtls.tof_set_params(role, samples_per_burst,
tof_num_freq, auto_tof_rssi,
tof_sample_mode, tof_run_mode,
tof_freq_list)
if aoa_supported:
# Find the role based on capabilities of sending node
role = 'AOA_MASTER' if sending_node.capabilities.get('AOA_MASTER', False) else 'AOA_PASSIVE'
# Send AoA params
sending_node.rtls.aoa_set_params(role, aoa_run_mode,
aoa_cte_scan_ovs,
aoa_cte_offset,
aoa_cte_time)
else:
# If the connection failed, keep scanning
master_node.rtls.scan()
# Count the number of nodes that have ToF initialized
if msg.command == 'RTLS_CMD_TOF_SET_PARAMS' and msg.payload.status == 'RTLS_SUCCESS':
sending_node.tof_initialized = True
# If all nodes have responded then we are ready to move on
if all([n.tof_initialized for n in all_nodes]):
# Send request for seed to master
master_node.rtls.tof_get_sec_seed()
if msg.command == 'RTLS_CMD_AOA_SET_PARAMS' and msg.payload.status == 'RTLS_SUCCESS':
sending_node.aoa_initialized = True
if all([n.aoa_initialized for n in all_nodes]):
# Start AoA on the master and passive nodes
for node in all_nodes:
node.rtls.aoa_start(True)
# Wait for security seed
if msg.command == 'RTLS_CMD_TOF_GET_SEC_SEED' and msg.payload.seed is not 0:
seed = msg.payload.seed
for node in passive_nodes:
node.rtls.tof_set_sec_seed(seed)
# Wait until passives have security seed set and start ToF
if msg.command == 'RTLS_CMD_TOF_SET_SEC_SEED' and msg.payload.status == 'RTLS_SUCCESS':
sending_node.seed_initialized = True
if all([n.seed_initialized for n in passive_nodes]):
for node in passive_nodes:
node.rtls.tof_start(True)
# Passive must start well before Master does since it must "hear" the first ToF exchange
master_node.rtls.tof_start(True)
# Wait until passives have security seed set. Set calibration option
if msg.command == 'RTLS_CMD_TOF_ENABLE' and msg.payload.status == 'RTLS_SUCCESS':
# Only need to calibrate in distance mode
if tof_sample_mode == 'TOF_MODE_DIST':
if sending_node in passive_nodes:
for node in passive_nodes:
node.rtls.tof_calib(True, samplesPerFreq, calibDistance)
else:
master_node.rtls.tof_calib(True, samplesPerFreq, calibDistance)
# Saving I/Q samples into csv file
if msg.command == 'RTLS_CMD_AOA_RESULT_RAW':
payload = msg.payload
# Extract first sample index in this payload
offset = payload.offset
# If we have data, and offset is 0, we are done with one dump
if offset == 0 and len(dump_rows):
pkt_cnt += 1
# Make sure the samples are in order
dump_rows = sorted(dump_rows, key=lambda s: s.sample_idx)
# Write to file
for sample_row in dump_rows:
csv_writer.writerow(sample_row._asdict())
# Reset payload storage
dump_rows = []
# Stop script now if there was a limit configured
if pkt_limit is not None and pkt_cnt > pkt_limit:
break
# Save samples for writing when dump is complete
for sub_idx, sample in enumerate(payload.samples):
sample = SampleRow(pkt=pkt_cnt, sample_idx=offset + sub_idx, rssi=payload.rssi, ant_array=payload.antenna, channel=payload.channel, i=sample.i, q=sample.q)
dump_rows.append(sample)
except queue.Empty:
pass
finally:
outfile.flush()
outfile.close()
if manager:
manager.stop()
Thanks and Regards
Akansha
