mirror of
https://gitlab.eurecom.fr/oai/openairinterface5g.git
synced 2026-07-13 04:30:28 +00:00
data-recording v1.1: Major rewrite of the synchronized real-time data recording system.
Application (v1.1): - Generic N-node architecture replacing hardcoded gNB/UE structure - Node-based JSON config with per-node shared memory paths - Thread pool with barrier sync for multi-node recording - Frame/slot-based message grouping in read_and_store_tracer_messages() - Robust cleanup via atexit and signal handlers - Remove v1.0 legacy application Shared Memory Interface (new): - System V shared memory attach/detach/remove with ftok-based keys - Structured binary read/write for tracer control messages - Frame/slot-aware data reading with configurable timeout Config Interface: - JSON-based config with node definitions (id, type, shared_mem_config) - Per-node tracer message parsing and index resolution SigMF Interface: - Timestamp refactor: Unix epoch sec/nsec with nanosecond precision - Wireless parameter mapping via YAML-driven parameter map - System components metadata (TX/channel/RX) in SigMF annotations Sync Service: - Generic multi-node sync replacing 2-node sync - Frame-wrap-aware comparison (1024-frame wrap) Wireless Parameters Mapper (new): - YAML-driven OAI-to-SigMF parameter mapping - 5G NR metadata derivation (DMRS, modulation, link direction)
This commit is contained in:
abgaber
@@ -42,43 +42,59 @@
|
||||
"t_tracer_message_definition_file": "../T/T_messages.txt",
|
||||
"parameter_map_file": "config/wireless_link_parameter_map.yaml",
|
||||
"start_frame_number": 5,
|
||||
"base_station": {
|
||||
"requested_tracer_messages": [
|
||||
"GNB_PHY_UL_FD_PUSCH_IQ",
|
||||
"GNB_PHY_UL_FD_DMRS",
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_POS",
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_INTERPL",
|
||||
"GNB_PHY_UL_PAYLOAD_RX_BITS"
|
||||
],
|
||||
"meta_data":{
|
||||
"num_rx_antennas": 1,
|
||||
"tx_gain": 48.0,
|
||||
"rx_gain": 30.0,
|
||||
"hw_type": "USRP X410",
|
||||
"hw_subtype": "ZBX",
|
||||
"seid": "328AB35"
|
||||
}
|
||||
},
|
||||
"user_equipment": {
|
||||
"requested_tracer_messages": [
|
||||
"UE_PHY_UL_SCRAMBLED_TX_BITS",
|
||||
"UE_PHY_UL_PAYLOAD_TX_BITS"
|
||||
],
|
||||
"meta_data":{
|
||||
"num_tx_antennas": 1,
|
||||
"tx_gain": 48.0,
|
||||
"rx_gain": 40.0,
|
||||
"hw_type": "USRP X410",
|
||||
"hw_subtype": "ZBX",
|
||||
"seid": "323F75F"
|
||||
}
|
||||
},
|
||||
"common_meta_data": {
|
||||
"sample_rate": 61440000.0,
|
||||
"bandwidth": 40000000.0,
|
||||
"clock_reference": "external"
|
||||
},
|
||||
"tracer_service_baseStation_address": "127.0.0.1:2021",
|
||||
"tracer_service_userEquipment_address": "127.0.0.1:2023"
|
||||
"nodes": [
|
||||
{
|
||||
"id": "gnb_0",
|
||||
"type": "gnb",
|
||||
"requested_tracer_messages": [
|
||||
"GNB_PHY_UL_FD_PUSCH_IQ",
|
||||
"GNB_PHY_UL_FD_DMRS",
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_POS",
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_INTERPL",
|
||||
"GNB_PHY_UL_PAYLOAD_RX_BITS"],
|
||||
"tracer_service_address": "127.0.0.1:2021",
|
||||
"meta_data": {
|
||||
"num_tx_antennas": 1,
|
||||
"num_rx_antennas": 1,
|
||||
"tx_gain": 48.0,
|
||||
"rx_gain": 30.0,
|
||||
"hw_type": "USRP X410",
|
||||
"hw_subtype": "ZBX",
|
||||
"seid": "328AB35"
|
||||
},
|
||||
"shared_mem_config": {
|
||||
"read_path": "/tmp/gnb_app1",
|
||||
"write_path": "/tmp/gnb_app2",
|
||||
"project_id": 2335
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "ue_0",
|
||||
"type": "ue",
|
||||
"requested_tracer_messages": [
|
||||
"UE_PHY_UL_SCRAMBLED_TX_BITS",
|
||||
"UE_PHY_UL_PAYLOAD_TX_BITS"],
|
||||
"tracer_service_address": "127.0.0.1:2023",
|
||||
"meta_data": {
|
||||
"num_tx_antennas": 1,
|
||||
"num_rx_antennas": 1,
|
||||
"tx_gain": 48.0,
|
||||
"rx_gain": 30.0,
|
||||
"hw_type": "USRP X410",
|
||||
"hw_subtype": "ZBX",
|
||||
"seid": "323F75F"
|
||||
},
|
||||
"shared_mem_config": {
|
||||
"read_path": "/tmp/ue_app1",
|
||||
"write_path": "/tmp/ue_app2",
|
||||
"project_id": 2336
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,803 +0,0 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief main application of synchronized real-time data recording
|
||||
|
||||
import sysv_ipc as ipc
|
||||
import struct
|
||||
import time
|
||||
from datetime import datetime
|
||||
from termcolor import colored
|
||||
import numpy as np
|
||||
import json
|
||||
import concurrent.futures
|
||||
# from concurrent import futures
|
||||
import threading
|
||||
# import related functions
|
||||
from lib import sigmf_interface
|
||||
# import library functions
|
||||
from lib import sync_service
|
||||
from lib import data_recording_messages_def
|
||||
from lib import common_utils
|
||||
from lib import config_interface
|
||||
|
||||
DEBUG_WIRELESS_RECORDED_DATA = True
|
||||
DEBUG_BUFFER_READING = False
|
||||
|
||||
# globally applicable metadata
|
||||
global_info = {
|
||||
"author": "Abdo Gaber",
|
||||
"description": "Synchronized Real-Time Data Recording",
|
||||
"timestamp": 0,
|
||||
"collection_file_prefix": "data-collection", # collection file name prefix "deap-rx + str(...)"
|
||||
"collection_file": "", # Reserved to be created in the code: “data-collection_rec-0_TIME-STAMP”
|
||||
"datetime_offset": "", # datetime offset between current location and UTC/Zulu timezone
|
||||
# Example: "+01:00" for Berlin, Germany
|
||||
"save_config_data_recording_app_json": True,
|
||||
"waveform_generator": "5gnr_oai",
|
||||
"extensions": {},
|
||||
}
|
||||
|
||||
# Supported OAI Trace messages
|
||||
# UL receiver messages
|
||||
# gNB IQ Msgs: "GNB_PHY_UL_FD_PUSCH_IQ", "GNB_PHY_UL_FD_DMRS", "GNB_PHY_UL_FD_CHAN_EST_DMRS_POS",
|
||||
# "GNB_PHY_UL_FD_CHAN_EST_DMRS_INTERPL"
|
||||
# gNB BITS Msgs: "GNB_PHY_UL_PAYLOAD_RX_BITS"
|
||||
# UE BITS Msgs: "UE_PHY_UL_SCRAMBLED_TX_BITS", "UE_PHY_UL_PAYLOAD_TX_BITS"
|
||||
|
||||
supported_oai_tracer_messages = {
|
||||
# gNB messages
|
||||
"GNB_PHY_UL_FD_PUSCH_IQ": {
|
||||
"file_name_prefix": "rx-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Frequency-domain RX data",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_FD_DMRS": {
|
||||
"file_name_prefix": "tx-pilots-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Frequency-domain TX PUSCH DMRS data",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_POS": {
|
||||
"file_name_prefix": "raw-ce-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Frequency-domain raw channel estimates (at DMRS positions)",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_INTERPL": {
|
||||
"file_name_prefix": "raw-inter-ce-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Interpolcated Frequency-domain raw channel estimates",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_PAYLOAD_RX_BITS": {
|
||||
"file_name_prefix": "rx-payload-bits",
|
||||
"scope": "gNB",
|
||||
"description": "Received PUSCH payload bits",
|
||||
"serialization_scheme": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
# UE messages
|
||||
"UE_PHY_UL_SCRAMBLED_TX_BITS": {
|
||||
"file_name_prefix": "tx-scrambled-bits",
|
||||
"scope": "UE",
|
||||
"description": "Transmitted scrambled PUSCH bits",
|
||||
"serialization_scheme": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"UE_PHY_UL_PAYLOAD_TX_BITS": {
|
||||
"file_name_prefix": "tx-payload-bits",
|
||||
"scope": "UE",
|
||||
"description": "Transmitted PUSCH payload bits",
|
||||
"serialization_scheme": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
}
|
||||
# -------------------------------------------
|
||||
# System configuration: gNB
|
||||
project_id_gnb = 2335
|
||||
read_shm_path_gnb = "/tmp/gnb_app1"
|
||||
write_shm_path_gnb = "/tmp/gnb_app2"
|
||||
|
||||
# System configuration: UE
|
||||
project_id_ue = 2336
|
||||
read_shm_path_ue = "/tmp/ue_app1"
|
||||
write_shm_path_ue = "/tmp/ue_app2"
|
||||
|
||||
|
||||
# initialize shared memory
|
||||
def attach_shm(shm_path, project_id):
|
||||
key = ipc.ftok(shm_path, project_id)
|
||||
shm = ipc.SharedMemory(key, 0, 0)
|
||||
# I found if we do not attach ourselves
|
||||
# it will attach as ReadOnly.
|
||||
shm.attach(0, 0)
|
||||
return shm
|
||||
|
||||
|
||||
def detach_shm(shm):
|
||||
try:
|
||||
shm.detach()
|
||||
print("Shared memory detached successfully.")
|
||||
except ipc.ExistentialError:
|
||||
print("Shared memory segment does not exist.")
|
||||
|
||||
|
||||
def remove_shm(shm):
|
||||
try:
|
||||
shm.remove()
|
||||
print("Shared memory removed successfully.")
|
||||
except ipc.ExistentialError:
|
||||
print("Shared memory segment does not exist.")
|
||||
|
||||
|
||||
# check data if avalible in the shared memory
|
||||
def is_data_available_in_memory(shm, bufIdx, general_message_header_length, timeout=20):
|
||||
start_time = time.time()
|
||||
while True:
|
||||
buf = shm.read(bufIdx + general_message_header_length)
|
||||
n_bytes = sum(buf)
|
||||
print("Data Recording App: Waiting for Measurements!")
|
||||
if n_bytes > 0:
|
||||
print("There is data in memory, n_bytes: ", n_bytes)
|
||||
return True
|
||||
if (time.time() - start_time) > timeout:
|
||||
break
|
||||
time.sleep(1)
|
||||
return False
|
||||
|
||||
|
||||
# Read data from Shared memory based Data Conversion Service message structure
|
||||
def read_data_from_shm(shm, bufIdx, tracer_msgs_identities):
|
||||
# print buffer index
|
||||
if DEBUG_BUFFER_READING:
|
||||
print("Buffer Index: ", bufIdx)
|
||||
# get general message header list
|
||||
general_msg_header_list, general_message_header_length = data_recording_messages_def.get_general_msg_header_list()
|
||||
buf = shm.read(bufIdx + general_message_header_length)
|
||||
n_bytes = sum(buf)
|
||||
if n_bytes == 0:
|
||||
raise Exception('ERROR: No data available in memory')
|
||||
msg_id = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("msg_id")])[0]
|
||||
bufIdx += general_msg_header_list.get("msg_id")
|
||||
frame = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("frame")])[0]
|
||||
bufIdx += general_msg_header_list.get("frame")
|
||||
slot = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("slot")])[0]
|
||||
bufIdx += general_msg_header_list.get("slot")
|
||||
# get time stamp: yyyy mm dd hh mm ss msec
|
||||
nr_trace_time_stamp_yyymmdd = \
|
||||
struct.unpack('<i', buf[bufIdx:bufIdx+general_msg_header_list.get("datetime_yyyymmdd")])[0]
|
||||
bufIdx += general_msg_header_list.get("datetime_yyyymmdd")
|
||||
nr_trace_time_stamp_hhmmssmmm = \
|
||||
struct.unpack('<i', buf[bufIdx:bufIdx+general_msg_header_list.get("datetime_hhmmssmmm")])[0]
|
||||
bufIdx += general_msg_header_list.get("datetime_hhmmssmmm")
|
||||
time_stamp_milli_sec = str(nr_trace_time_stamp_yyymmdd)+"_"+str(nr_trace_time_stamp_hhmmssmmm)
|
||||
# get frame type
|
||||
frame_type = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("frame_type")])[0]
|
||||
bufIdx += general_msg_header_list.get("frame_type")
|
||||
# get frequency range
|
||||
freq_range = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("freq_range")])[0]
|
||||
bufIdx += general_msg_header_list.get("freq_range")
|
||||
# get subcarrier spacing
|
||||
subcarrier_spacing = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("subcarrier_spacing")])[0]
|
||||
bufIdx += general_msg_header_list.get("subcarrier_spacing")
|
||||
# get cyclic prefix
|
||||
cyclic_prefix = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("cyclic_prefix")])[0]
|
||||
bufIdx += general_msg_header_list.get("cyclic_prefix")
|
||||
# get symbols per slot
|
||||
symbols_per_slot = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("symbols_per_slot")])[0]
|
||||
bufIdx += general_msg_header_list.get("symbols_per_slot")
|
||||
# get Nid cell
|
||||
Nid_cell = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("Nid_cell")])[0]
|
||||
bufIdx += general_msg_header_list.get("Nid_cell")
|
||||
# get rnti
|
||||
rnti = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("rnti")])[0]
|
||||
bufIdx += general_msg_header_list.get("rnti")
|
||||
# get rb size
|
||||
rb_size = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("rb_size")])[0]
|
||||
bufIdx += general_msg_header_list.get("rb_size")
|
||||
# get rb start
|
||||
rb_start = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("rb_start")])[0]
|
||||
bufIdx += general_msg_header_list.get("rb_start")
|
||||
# get start symbol index
|
||||
start_symbol_index = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("start_symbol_index")])[0]
|
||||
bufIdx += general_msg_header_list.get("start_symbol_index")
|
||||
# get number of symbols
|
||||
nr_of_symbols = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("nr_of_symbols")])[0]
|
||||
bufIdx += general_msg_header_list.get("nr_of_symbols")
|
||||
# get qam modulation order
|
||||
qam_mod_order = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("qam_mod_order")])[0]
|
||||
bufIdx += general_msg_header_list.get("qam_mod_order")
|
||||
# get mcs index
|
||||
mcs_index = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("mcs_index")])[0]
|
||||
bufIdx += general_msg_header_list.get("mcs_index")
|
||||
# get mcs table
|
||||
mcs_table = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("mcs_table")])[0]
|
||||
bufIdx += general_msg_header_list.get("mcs_table")
|
||||
# get number of layers
|
||||
nrOfLayers = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("nrOfLayers")])[0]
|
||||
bufIdx += general_msg_header_list.get("nrOfLayers")
|
||||
# get transform precoding
|
||||
transform_precoding = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("transform_precoding")])[0]
|
||||
bufIdx += general_msg_header_list.get("transform_precoding")
|
||||
# get dmrs config type
|
||||
dmrs_config_type = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("dmrs_config_type")])[0]
|
||||
bufIdx += general_msg_header_list.get("dmrs_config_type")
|
||||
# get ul dmrs symb pos
|
||||
ul_dmrs_symb_pos = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("ul_dmrs_symb_pos")])[0]
|
||||
bufIdx += general_msg_header_list.get("ul_dmrs_symb_pos")
|
||||
# get number dmrs symbols
|
||||
number_dmrs_symbols = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("number_dmrs_symbols")])[0]
|
||||
bufIdx += general_msg_header_list.get("number_dmrs_symbols")
|
||||
# get dmrs port
|
||||
dmrs_port = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("dmrs_port")])[0]
|
||||
bufIdx += general_msg_header_list.get("dmrs_port")
|
||||
# get dmrs scid
|
||||
dmrs_scid = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("dmrs_scid")])[0]
|
||||
bufIdx += general_msg_header_list.get("dmrs_scid")
|
||||
# get nb antennas rx for gNB or nb antennas tx for UE
|
||||
nb_antennas = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("nb_antennas")])[0]
|
||||
bufIdx += general_msg_header_list.get("nb_antennas")
|
||||
# get number of bits
|
||||
number_of_bits = struct.unpack('<I', buf[bufIdx:bufIdx+general_msg_header_list.get("number_of_bits")])[0]
|
||||
bufIdx += general_msg_header_list.get("number_of_bits")
|
||||
# get length of bytes
|
||||
length_bytes = struct.unpack('<I', buf[bufIdx:bufIdx+general_msg_header_list.get("length_bytes")])[0]
|
||||
bufIdx += general_msg_header_list.get("length_bytes")
|
||||
|
||||
# print all captured data
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(" ")
|
||||
print(f"Time stamp: {time_stamp_milli_sec}")
|
||||
print(f"MSG ID: {msg_id:<5} MSG Name: {tracer_msgs_identities[msg_id]}")
|
||||
print(f"Frame: {frame:<5} Slot: {slot:<5}")
|
||||
print(f"Frame Type: {frame_type:<5} Frequency Range: {freq_range:<5}"
|
||||
f"Subcarrier Spacing: {subcarrier_spacing:<5} Cyclic Prefix: {cyclic_prefix:<5} "
|
||||
f"Symbols per Slot: {symbols_per_slot:<5}")
|
||||
print(f"Nid Cell: {Nid_cell:<5} RNTI: {rnti:<5}")
|
||||
print(f"RB Size: {rb_size:<5} RB Start: {rb_start:<5} Start Symbol Index: {start_symbol_index:<5} "
|
||||
f"Number of Symbols: {nr_of_symbols:<5}")
|
||||
print(f"QAM Modulation Order: {qam_mod_order:<5} MCS Index: {mcs_index:<5} "
|
||||
f"MCS Table: {mcs_table:<5}")
|
||||
print(f"Number of Layers: {nrOfLayers:<5} Transform Precoding: {transform_precoding:<5}")
|
||||
print(f"DMRS Config Type: {dmrs_config_type:<5} UL DMRS Symbol Position: {ul_dmrs_symb_pos:<5} "
|
||||
f"Number of DMRS Symbols: {number_dmrs_symbols:<5}")
|
||||
print(f"DMRS Port: {dmrs_port:<5} DMRS SCID: {dmrs_scid:<5} "
|
||||
f"Number of Antennas: {nb_antennas:<5}")
|
||||
print(f"Number of bits: {number_of_bits:<5} Length of bytes: {length_bytes:<5}")
|
||||
# raise exception if time stamp is zero, it means that the data is not recorded yet
|
||||
if nr_trace_time_stamp_yyymmdd == 0 and nr_trace_time_stamp_hhmmssmmm == 0:
|
||||
raise Exception("ERROR: Time stamp is zero, data is not recorded yet or something wrong, check logs!")
|
||||
# get recorded data
|
||||
buf = shm.read(bufIdx + length_bytes)
|
||||
# bit_msg_index = get_index_of_id(tracer_msgs_identities, "GNB_PHY_UL_PAYLOAD_RX_BITS")
|
||||
captured_data = {}
|
||||
# If message is bit message, store data in bytes
|
||||
# then the field number_of_bits should be not zero
|
||||
if "_BITS" in tracer_msgs_identities[msg_id]:
|
||||
# recorded_data = buf[bufIdx:bufIdx + length_bytes]
|
||||
recorded_data = struct.unpack("<" + int(length_bytes) * 'B', buf[bufIdx:bufIdx + length_bytes])
|
||||
bufIdx += length_bytes
|
||||
# convert data in bytes to bits
|
||||
bits_vector = []
|
||||
for byte in recorded_data:
|
||||
bits_vector.extend([int(bit) for bit in format(int(byte), '08b')])
|
||||
captured_data["sigmf_data_type"] = "ri8_le"
|
||||
# convert to uint8
|
||||
captured_data["recorded_data"] = np.asarray(bits_vector).astype(np.uint8)
|
||||
# recorded_data_formated = recorded_data.astype(np.complex64) # convert to complex64
|
||||
else:
|
||||
recorded_data = struct.unpack("<" + int(length_bytes/2) * 'h', buf[bufIdx:bufIdx + length_bytes])
|
||||
bufIdx += length_bytes
|
||||
# print("IQ data I/Q: ", recorded_data)
|
||||
# Convert real data to complext data
|
||||
# converting list to array
|
||||
recorded_data = np.asarray(recorded_data)
|
||||
# recorded_data_complex = recorded_data
|
||||
recorded_data_complex = common_utils.real_to_complex(recorded_data)
|
||||
captured_data["sigmf_data_type"] = "cf32_le"
|
||||
# convert to complex64
|
||||
captured_data["recorded_data"] = recorded_data_complex.astype(np.complex64)
|
||||
# print("Recorded Data: ", captured_data["recorded_data"])
|
||||
# store data in dictonary
|
||||
captured_data["message_id"] = msg_id
|
||||
captured_data["message_type"] = tracer_msgs_identities[msg_id]
|
||||
captured_data["frame"] = frame
|
||||
captured_data["slot"] = slot
|
||||
captured_data["time_stamp"] = time_stamp_milli_sec
|
||||
captured_data["frame_type"] = frame_type
|
||||
captured_data["freq_range"] = freq_range
|
||||
captured_data["subcarrier_spacing"] = subcarrier_spacing
|
||||
captured_data["cyclic_prefix"] = cyclic_prefix
|
||||
# captured_data["symbols_per_slot"] = symbols_per_slot ... not used
|
||||
captured_data["Nid_cell"] = Nid_cell
|
||||
captured_data["rnti"] = rnti
|
||||
captured_data["rb_size"] = rb_size
|
||||
captured_data["rb_start"] = rb_start
|
||||
captured_data["start_symbol_index"] = start_symbol_index
|
||||
captured_data["nr_of_symbols"] = nr_of_symbols
|
||||
captured_data["qam_mod_order"] = qam_mod_order
|
||||
captured_data["mcs_index"] = mcs_index
|
||||
captured_data["mcs_table"] = mcs_table
|
||||
captured_data["nrOfLayers"] = nrOfLayers
|
||||
captured_data["transform_precoding"] = transform_precoding
|
||||
captured_data["dmrs_config_type"] = dmrs_config_type
|
||||
captured_data["ul_dmrs_symb_pos"] = ul_dmrs_symb_pos
|
||||
captured_data["number_dmrs_symbols"] = number_dmrs_symbols
|
||||
captured_data["dmrs_port"] = dmrs_port
|
||||
captured_data["dmrs_scid"] = dmrs_scid
|
||||
captured_data["nb_antennas"] = nb_antennas
|
||||
captured_data["number_of_bits"] = number_of_bits
|
||||
return captured_data, bufIdx
|
||||
|
||||
|
||||
# Synchronize data between gNB and UE
|
||||
def sync_data_conversion_service(
|
||||
shm_reading_gnb, shm_reading_ue, sync_info, config_meta_data, gnb_args, ue_args):
|
||||
# Initialize variables
|
||||
record_idx = 0
|
||||
prev_frame = -1
|
||||
prev_slot = -1
|
||||
ue_bufIdx = 0
|
||||
gnb_bufIdx = 0
|
||||
|
||||
gnb_args.num_requested_tracer_msgs = len(
|
||||
config_meta_data["data_recording_config"]["base_station"][
|
||||
"requested_tracer_messages"])
|
||||
ue_args.num_requested_tracer_msgs = len(
|
||||
config_meta_data["data_recording_config"]["user_equipment"][
|
||||
"requested_tracer_messages"])
|
||||
tracer_msgs_identities = config_meta_data["data_recording_config"][
|
||||
"tracer_msgs_identities"]
|
||||
|
||||
global_info = config_meta_data["data_recording_config"]["global_info"]
|
||||
|
||||
# Get UE data based on the sync data
|
||||
bufIdx = 0
|
||||
timeout_sync = time.time() + 5 # 5 seconds if no sync data found, stop the process
|
||||
while True:
|
||||
# wait for the next record
|
||||
# To do: check if we need to add exta waiting times between different events in case of
|
||||
# data streaming via network such as on UE side or gNB side
|
||||
time.sleep(0.0035) # 2.3 ms = latency of T tracer to capture data from the RAN
|
||||
ue_bufIdx = bufIdx
|
||||
captured_data, bufIdx = read_data_from_shm(shm_reading_ue, bufIdx, tracer_msgs_identities)
|
||||
if (captured_data["frame"] == sync_info["frame"]
|
||||
and captured_data["slot"] == sync_info["slot"]):
|
||||
break
|
||||
if time.time() > timeout_sync:
|
||||
raise Exception(
|
||||
"ERROR: Data Recording NO Sync Found, check Tracer Services if they are connected!")
|
||||
|
||||
# Get gNB data based on the sync data
|
||||
bufIdx = 0
|
||||
while True:
|
||||
time.sleep(0.0035)
|
||||
gnb_bufIdx = bufIdx
|
||||
captured_data, bufIdx = read_data_from_shm(
|
||||
shm_reading_gnb, bufIdx, tracer_msgs_identities)
|
||||
if (captured_data["frame"] == sync_info["frame"]
|
||||
and captured_data["slot"] == sync_info["slot"]):
|
||||
break
|
||||
|
||||
# Read Synchronized data between gNB and UE
|
||||
while True: # read all records
|
||||
print("\nRecord number: ", record_idx)
|
||||
if DEBUG_BUFFER_READING:
|
||||
print(f"Buffer Index gNB: {gnb_bufIdx}, Buffer Index UE: {ue_bufIdx}")
|
||||
# wait for the next record
|
||||
# To do: check if we need to add exta waiting times between different events in case of
|
||||
# data streaming via network such as on UE side or gNB side
|
||||
time.sleep(0.0035) # 2.3 ms = latency of T tracer to capture data from the RAN
|
||||
|
||||
collected_metafiles = []
|
||||
# Read data from gNB T-tracer Application
|
||||
for idx in range(gnb_args.num_requested_tracer_msgs):
|
||||
time.sleep(0.0015)
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"\nRecord number: {record_idx}, Reading MSG data ", idx)
|
||||
captured_data, gnb_bufIdx = read_data_from_shm(
|
||||
shm_reading_gnb, gnb_bufIdx, tracer_msgs_identities)
|
||||
# drive the collection file time stamp from the first message per record
|
||||
if idx == 0:
|
||||
# Get time stamp
|
||||
time_stamp_ms, time_stamp_ms_file_name = (
|
||||
sigmf_interface.time_stamp_formating(
|
||||
captured_data["time_stamp"], global_info["datetime_offset"]))
|
||||
global_info["collection_file"] = (
|
||||
global_info["collection_file_prefix"]
|
||||
+ "-rec-"
|
||||
+ str(record_idx)
|
||||
+ "-"
|
||||
+ str(time_stamp_ms_file_name)
|
||||
)
|
||||
global_info["timestamp"] = time_stamp_ms
|
||||
|
||||
# Write data into files with the given format
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
collected_metafiles.append(
|
||||
sigmf_interface.write_recorded_data_to_sigmf(
|
||||
captured_data, config_meta_data, global_info, record_idx))
|
||||
|
||||
# Read data from UE T-tracer Application
|
||||
for idx in range(ue_args.num_requested_tracer_msgs):
|
||||
time.sleep(0.0015)
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"\nRecord number: {record_idx}, Reading MSG data ", idx)
|
||||
captured_data, ue_bufIdx = read_data_from_shm(
|
||||
shm_reading_ue, ue_bufIdx, tracer_msgs_identities)
|
||||
|
||||
# Write data into files with the given format
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
collected_metafiles.append(
|
||||
sigmf_interface.write_recorded_data_to_sigmf(
|
||||
captured_data, config_meta_data, global_info, record_idx))
|
||||
|
||||
# generate SigMF collection file
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
data_storage_path = config_meta_data["data_recording_config"][
|
||||
"data_storage_path"]
|
||||
description = global_info["description"]
|
||||
sigmf_interface.save_sigmf_collection(
|
||||
collected_metafiles, global_info, description, data_storage_path)
|
||||
|
||||
frame = captured_data["frame"]
|
||||
slot = captured_data["slot"]
|
||||
|
||||
# Check for changes in frame or slot
|
||||
if frame != prev_frame or slot != prev_slot:
|
||||
record_idx += 1
|
||||
# We have reached the end of the data. Break the loop
|
||||
if record_idx >= config_meta_data["data_recording_config"]["num_records"]:
|
||||
break
|
||||
# Update previous frame and slot
|
||||
prev_frame = frame
|
||||
prev_slot = slot
|
||||
|
||||
|
||||
# data conversion service
|
||||
def data_conversion_service(shm_reading, config_meta_data, args, sync_info, do_sync):
|
||||
# Initialize variables
|
||||
record_idx = 0
|
||||
prev_frame = -1
|
||||
prev_slot = -1
|
||||
bufIdx = 0
|
||||
# Read data from T-tracer Application
|
||||
print("Data Conversion Service: Reading data from T-tracer Application")
|
||||
print("Requested Tracer Messages: ", args.num_requested_tracer_msgs)
|
||||
if args.num_requested_tracer_msgs > 0:
|
||||
num_requested_tracer_msgs = args.num_requested_tracer_msgs
|
||||
else:
|
||||
raise Exception("ERROR: No requested tracer messages found!")
|
||||
|
||||
tracer_msgs_identities = config_meta_data["data_recording_config"][
|
||||
"tracer_msgs_identities"]
|
||||
global_info = config_meta_data["data_recording_config"]["global_info"]
|
||||
|
||||
if do_sync:
|
||||
print(" Find Memory Index of NR MSGs based on Sync info")
|
||||
while True:
|
||||
time.sleep(0.0035)
|
||||
station_bufIdx = bufIdx
|
||||
captured_data, bufIdx = read_data_from_shm(
|
||||
shm_reading, bufIdx, tracer_msgs_identities)
|
||||
print("*Sync Status: NR Captured Data (Frame, slot): (", captured_data["frame"],
|
||||
", ", captured_data["slot"], "), Sync Info (Frame, slot): (", sync_info["frame"],
|
||||
", ", sync_info["slot"]," )")
|
||||
if (captured_data["frame"] == sync_info["frame"]
|
||||
and captured_data["slot"] == sync_info["slot"]):
|
||||
print("*sync Pass")
|
||||
break
|
||||
print("*sync Fail")
|
||||
bufIdx = station_bufIdx
|
||||
|
||||
# Read data from T-tracer Application
|
||||
while True: # read all records
|
||||
print("\nRecord number: ", record_idx)
|
||||
if DEBUG_BUFFER_READING:
|
||||
print(f"Buffer Index: {bufIdx}")
|
||||
# wait for the next record
|
||||
# To do: check if we need to add exta waiting times between different events in case of
|
||||
# data streaming via network such as on UE side or gNB side
|
||||
time.sleep(0.0035) # 2.3 ms = latency of T tracer to capture data from the RAN
|
||||
|
||||
collected_metafiles = []
|
||||
for idx in range(num_requested_tracer_msgs):
|
||||
time.sleep(0.0015)
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"\nRecord number: {record_idx}, Reading MSG data ", idx)
|
||||
captured_data, bufIdx = read_data_from_shm(
|
||||
shm_reading, bufIdx, tracer_msgs_identities)
|
||||
# derive the collection file time stamp from the first message per record
|
||||
if idx == 0:
|
||||
# Get time stamp
|
||||
time_stamp_ms, time_stamp_ms_file_name = (
|
||||
sigmf_interface.time_stamp_formating(
|
||||
captured_data["time_stamp"], global_info["datetime_offset"]))
|
||||
global_info["collection_file"] = (
|
||||
global_info["collection_file_prefix"]
|
||||
+ "-rec-"
|
||||
+ str(record_idx)
|
||||
+ "-"
|
||||
+ str(time_stamp_ms_file_name))
|
||||
global_info["timestamp"] = time_stamp_ms
|
||||
# Write data into files with the given format
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
collected_metafiles.append(
|
||||
sigmf_interface.write_recorded_data_to_sigmf(
|
||||
captured_data, config_meta_data, global_info, record_idx))
|
||||
|
||||
frame = captured_data["frame"]
|
||||
slot = captured_data["slot"]
|
||||
|
||||
# generate SigMF collection file
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
data_storage_path = config_meta_data["data_recording_config"][
|
||||
"data_storage_path"]
|
||||
description = global_info["description"]
|
||||
sigmf_interface.save_sigmf_collection(
|
||||
collected_metafiles, global_info, description, data_storage_path)
|
||||
|
||||
# Check for changes in frame or slot
|
||||
if frame != prev_frame or slot != prev_slot:
|
||||
record_idx += 1
|
||||
# We have reached the end of the data. Break the loop
|
||||
if record_idx >= config_meta_data["data_recording_config"]["num_records"]:
|
||||
break
|
||||
# Update previous frame and slot
|
||||
prev_frame = frame
|
||||
prev_slot = slot
|
||||
|
||||
|
||||
# Write Tracer Control Message
|
||||
def write_shm(shm, args):
|
||||
# Note: Big Endian >, Little Endian <
|
||||
# Note: unsigned char B, signed char b, short h, int I, long long q
|
||||
# Note: float f, double d, string s, char c, bool ?
|
||||
# 1: config
|
||||
# 2: record
|
||||
# 3: quit
|
||||
print("Write Shared Memory: ", args.action)
|
||||
if args.action == "config":
|
||||
# Determine the length of the IP address
|
||||
ip_length = len(args.bytes_IPaddress) + 1 # String terminator
|
||||
# Construct the format string dynamically
|
||||
format_string = f"{ip_length}s"
|
||||
shm.write(
|
||||
# Config action
|
||||
struct.pack("<B", 1) +
|
||||
struct.pack("<B", ip_length) +
|
||||
struct.pack(format_string, args.bytes_IPaddress) +
|
||||
struct.pack("<h", int(args.port))
|
||||
)
|
||||
print("T-Tracer Config IP: ", args.bytes_IPaddress, " port: ", args.port)
|
||||
elif args.action == "record":
|
||||
shm.write(
|
||||
# Record action
|
||||
struct.pack("<B", int(2)) +
|
||||
struct.pack("<B", args.num_requested_tracer_msgs) +
|
||||
struct.pack(
|
||||
"<{}h".format(len(args.req_tracer_msgs_indices)),
|
||||
*args.req_tracer_msgs_indices,) +
|
||||
struct.pack("<I", args.num_records) +
|
||||
struct.pack("<h", args.start_frame_number)
|
||||
)
|
||||
print("T-Tracer Record: N Messags: ", args.num_requested_tracer_msgs,
|
||||
", Msg IDs: ", args.req_tracer_msgs_indices,
|
||||
", Num records: ", args.num_records,
|
||||
", Start Frame: ", args.start_frame_number,
|
||||
)
|
||||
elif args.action == "quit":
|
||||
shm.write(struct.pack("<B", int(3))) # Quit action
|
||||
print("T-Tracer Quit")
|
||||
else:
|
||||
print("Unknown action for data recording system!")
|
||||
|
||||
|
||||
# write shared memory task
|
||||
def write_shm_task(barrier, shm_id, args):
|
||||
# Wait for threads to be ready
|
||||
barrier.wait()
|
||||
# send request to related T-Tracer Application
|
||||
write_shm(shm_id, args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# -------------------------------------------
|
||||
# ------------- Configuration --------------
|
||||
# ------------------------------------------
|
||||
# Data Recording Configuration
|
||||
data_recording_config_file = "config/config_data_recording.json"
|
||||
# -------------------------------------------
|
||||
# Configuration
|
||||
# -------------------------------------------
|
||||
# First: get the configuration mode either local or remote
|
||||
# Second: get data recording configuration
|
||||
# Read and parse the JSON file
|
||||
with open(data_recording_config_file, "r") as file:
|
||||
config_meta_data = json.load(file)
|
||||
|
||||
# get Configuration parameters
|
||||
config_meta_data, gnb_args, ue_args = config_interface.get_data_recording_config(config_meta_data)
|
||||
|
||||
# check if lists of requested tracer messages IDs are not empty
|
||||
if not gnb_args.requested_tracer_messages and \
|
||||
not ue_args.requested_tracer_messages:
|
||||
raise Exception("ERROR: No requested tracer messages are provided")
|
||||
|
||||
# check if gnb_requested_tracer_messages is not empty, attach to the shared memory
|
||||
if gnb_args.requested_tracer_messages:
|
||||
# attach to the shared memory
|
||||
shm_writing_gnb = attach_shm(write_shm_path_gnb, project_id_gnb)
|
||||
shm_reading_gnb = attach_shm(read_shm_path_gnb, project_id_gnb)
|
||||
|
||||
# check if ue_requested_tracer_messages is not empty, attach to the shared memory
|
||||
if ue_args.requested_tracer_messages:
|
||||
# attach to the shared memory
|
||||
shm_writing_ue = attach_shm(write_shm_path_ue, project_id_ue)
|
||||
shm_reading_ue = attach_shm(read_shm_path_ue, project_id_ue)
|
||||
|
||||
# get general message header list
|
||||
general_msg_header_list, general_message_header_length = \
|
||||
data_recording_messages_def.get_general_msg_header_list()
|
||||
|
||||
# Add supported OAI Tracer Messages
|
||||
config_meta_data["data_recording_config"]["supported_oai_tracer_messages"] = supported_oai_tracer_messages
|
||||
# Add global info
|
||||
config_meta_data["data_recording_config"]["global_info"] = global_info
|
||||
|
||||
# -------------------------------------------
|
||||
# Initialization
|
||||
# -------------------------------------------
|
||||
# -------------------------------------------
|
||||
# send Tracer Control Message request to T-Tracers Apps
|
||||
# -------------------------------------------
|
||||
# It consists of the following fields:
|
||||
# Config action
|
||||
# IP address length
|
||||
# IP address
|
||||
# Port Number
|
||||
# check if gnb_requested_tracer_messages is not empty, config T-Tracer gNB via shared memory
|
||||
if gnb_args.requested_tracer_messages:
|
||||
# Config T-Tracer via shared memory
|
||||
gnb_args.action = "config"
|
||||
write_shm(shm_writing_gnb, gnb_args)
|
||||
|
||||
# check if ue_requested_tracer_messages is not empty, config T-Tracer UE via shared memory
|
||||
if ue_args.requested_tracer_messages:
|
||||
# Config T-Tracer via shared memory
|
||||
ue_args.action = "config"
|
||||
write_shm(shm_writing_ue, ue_args)
|
||||
|
||||
time.sleep(0.5) # wait for the config to be applied
|
||||
|
||||
# -------------------------------------------
|
||||
# Execution
|
||||
# -------------------------------------------
|
||||
# send Tracer Control Message request to T-Tracers Apps
|
||||
# -------------------------------------------
|
||||
# It consists of the following fields:
|
||||
# Record action
|
||||
# Number of requested Tracer Messages
|
||||
# Requested Tracer Messages ID 1, …, ID N
|
||||
# Number of records to be recorded in slots
|
||||
# Start SFN: Frame Index to start data collection from it, useful for future
|
||||
# data sync between gNB and UE but not yet used
|
||||
print("Args:")
|
||||
if gnb_args.requested_tracer_messages:
|
||||
gnb_args.action = "record"
|
||||
print("gnb_args: ", gnb_args)
|
||||
if ue_args.requested_tracer_messages:
|
||||
ue_args.action = "record"
|
||||
print("ue_args: ", ue_args)
|
||||
|
||||
start_time = time.time()
|
||||
print("Send data logging request us:", datetime.now().strftime("%Y%m%d-%H%M%S%f"))
|
||||
|
||||
# if requested: gNB and UE Tracer Messages
|
||||
# gNB + UE Tracer Messages
|
||||
if gnb_args.requested_tracer_messages and ue_args.requested_tracer_messages:
|
||||
# Create a barrier to synchronize the threads
|
||||
barrier = threading.Barrier(2)
|
||||
with concurrent.futures.ThreadPoolExecutor() as executor:
|
||||
tracer_ue = executor.submit(write_shm_task, barrier, shm_writing_ue, ue_args)
|
||||
tracer_gnb = executor.submit(write_shm_task, barrier, shm_writing_gnb, gnb_args)
|
||||
# Wait for both functions to complete
|
||||
concurrent.futures.wait([tracer_ue, tracer_gnb])
|
||||
|
||||
# gNB Tracer Messages
|
||||
elif gnb_args.requested_tracer_messages:
|
||||
write_shm(shm_writing_gnb, gnb_args)
|
||||
|
||||
# UE Tracer Messages
|
||||
elif ue_args.requested_tracer_messages:
|
||||
write_shm(shm_writing_ue, ue_args)
|
||||
else:
|
||||
raise Exception("ERROR: No requested tracer messages IDs are provided")
|
||||
|
||||
# -------------------------------------------
|
||||
# Read data from gNB and UE T-tracer Application
|
||||
# -------------------------------------------
|
||||
# Check if data is available in memory
|
||||
# Initialize variables
|
||||
bufIdx = 0
|
||||
timeout = 10 # 10 seconds from now
|
||||
|
||||
# If gNB MSGs are requested
|
||||
if gnb_args.requested_tracer_messages:
|
||||
# Check if data is available in gNB memory
|
||||
is_gnb_data_in_memory = is_data_available_in_memory(
|
||||
shm_reading_gnb, bufIdx, general_message_header_length, timeout)
|
||||
# Report the status of gNB T-Tracer APP locally
|
||||
if not is_gnb_data_in_memory:
|
||||
print("Error: gNB: Check t-Tracer APP of gNB, check IPs and Ports")
|
||||
print("Error: gNB: If IPs and Ports are correct, re-run the hanging app.")
|
||||
print("It seems the socket was not closed properly")
|
||||
raise Exception("ERROR: Time out, check if gNB T-Tracer APP connected to stack")
|
||||
# If UE MSGs are requested
|
||||
if ue_args.requested_tracer_messages:
|
||||
# Check if data is available in UE memory
|
||||
is_ue_data_in_memory = is_data_available_in_memory(
|
||||
shm_reading_ue, bufIdx, general_message_header_length, timeout)
|
||||
# Report the status of UE T-Tracer APP locally
|
||||
if not is_ue_data_in_memory:
|
||||
print("Error: UE: Check t-Tracer APP of UE, check IPs and Ports")
|
||||
print("Error: UE: If IPs and Ports are correct, re-run the hanging app.")
|
||||
print("It seems the socket was not closed properly")
|
||||
raise Exception("ERROR: Time out, check if UE T-Tracer APP connected to stack")
|
||||
|
||||
# -------------------------------------------
|
||||
# Sync data between gNB and UE
|
||||
# -------------------------------------------
|
||||
# write JSON file
|
||||
common_utils.write_config_data_recording_app_json(config_meta_data)
|
||||
sync_info = {}
|
||||
if gnb_args.requested_tracer_messages and ue_args.requested_tracer_messages:
|
||||
# Sync data between gNB and UE
|
||||
sync_info = sync_service.sync_gnb_ue_captured_data(shm_reading_gnb, shm_reading_ue)
|
||||
print("\n***Sync data between gNB and UE: ", sync_info)
|
||||
# Read data from gNB and UE T-tracer Applications
|
||||
sync_data_conversion_service(
|
||||
shm_reading_gnb, shm_reading_ue, sync_info, config_meta_data, gnb_args, ue_args)
|
||||
elif gnb_args.requested_tracer_messages:
|
||||
# Read data from gNB T-tracer Application
|
||||
data_conversion_service(
|
||||
shm_reading_gnb, config_meta_data, gnb_args, sync_info, do_sync=False)
|
||||
elif ue_args.requested_tracer_messages:
|
||||
# Read data from UE T-tracer Application
|
||||
data_conversion_service(
|
||||
shm_reading_ue, config_meta_data, ue_args, sync_info, do_sync=False)
|
||||
else:
|
||||
raise Exception("ERROR: No requested tracer messages IDs are provided")
|
||||
|
||||
# measure Elapsed time
|
||||
time_elapsed = time.time() - start_time
|
||||
time_elapsed_ms = int(time_elapsed * 1000)
|
||||
print(
|
||||
"Elapsed time of getting Requested Messages and writing data and meta data files:",
|
||||
colored(time_elapsed_ms, "yellow"), "ms",)
|
||||
|
||||
# Stop T-Tracer Application function
|
||||
if gnb_args.requested_tracer_messages:
|
||||
gnb_args.action = "quit"
|
||||
write_shm(shm_writing_gnb, gnb_args)
|
||||
# Add Sleep time to ensure that the message sent to the UE T-tracer application is received
|
||||
# before the shared memory is detached
|
||||
time.sleep(0.5)
|
||||
# Clean shared memory
|
||||
detach_shm(shm_reading_gnb)
|
||||
detach_shm(shm_writing_gnb)
|
||||
remove_shm(shm_reading_gnb)
|
||||
remove_shm(shm_writing_gnb)
|
||||
if ue_args.requested_tracer_messages:
|
||||
ue_args.action = "quit"
|
||||
write_shm(shm_writing_ue, ue_args)
|
||||
# Add Sleep time to ensure that the message sent to the UE T-tracer application is received
|
||||
# before the shared memory is detached
|
||||
time.sleep(0.5)
|
||||
# Clean shared memory
|
||||
detach_shm(shm_reading_ue)
|
||||
detach_shm(shm_writing_ue)
|
||||
remove_shm(shm_reading_ue)
|
||||
remove_shm(shm_writing_ue)
|
||||
|
||||
print("End of the RF Data Recording API")
|
||||
pass
|
||||
523
common/utils/data_recording/data_recording_app_v1.1.py
Normal file
523
common/utils/data_recording/data_recording_app_v1.1.py
Normal file
@@ -0,0 +1,523 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief main application of synchronized real-time data recording
|
||||
|
||||
import atexit
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime
|
||||
import json
|
||||
import concurrent.futures
|
||||
from termcolor import colored
|
||||
import threading
|
||||
# import library functions
|
||||
from lib import sigmf_interface
|
||||
from lib import sync_service
|
||||
from lib import data_recording_messages_def
|
||||
from lib import common_utils
|
||||
from lib import config_interface
|
||||
from lib import shared_memory_interface as shm_interface
|
||||
|
||||
DEBUG_WIRELESS_RECORDED_DATA = True
|
||||
DEBUG_BUFFER_READING = False
|
||||
|
||||
# globally applicable metadata
|
||||
global_info = {
|
||||
"author": "Abdo Gaber",
|
||||
"description": "Synchronized Real-Time Data Recording",
|
||||
"timestamp": 0,
|
||||
"collection_file_prefix": "data-collection", # collection file name prefix
|
||||
"collection_file": "", # Reserved to be created in the code: “data-collection_rec-0_TIME-STAMP”
|
||||
"datetime_offset": "", # datetime offset between current location and UTC/Zulu timezone
|
||||
# Example: "+01:00" for Berlin, Germany
|
||||
"save_config_data_recording_app_json": True,
|
||||
"waveform_generator": "5gnr_oai",
|
||||
"extensions": {},
|
||||
}
|
||||
|
||||
# Supported OAI Trace messages
|
||||
# UL receiver messages
|
||||
# gNB IQ Msgs: "GNB_PHY_UL_FD_PUSCH_IQ", "GNB_PHY_UL_FD_DMRS", "GNB_PHY_UL_FD_CHAN_EST_DMRS_POS",
|
||||
# "GNB_PHY_UL_FD_CHAN_EST_DMRS_INTERPL"
|
||||
# gNB BITS Msgs: "GNB_PHY_UL_PAYLOAD_RX_BITS"
|
||||
# UE BITS Msgs: "UE_PHY_UL_SCRAMBLED_TX_BITS", "UE_PHY_UL_PAYLOAD_TX_BITS"
|
||||
|
||||
supported_oai_tracer_messages = {
|
||||
# gNB messages
|
||||
"GNB_PHY_UL_FD_PUSCH_IQ": {
|
||||
"file_name_prefix": "rx-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Frequency-domain RX data",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_FD_DMRS": {
|
||||
"file_name_prefix": "tx-pilots-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Frequency-domain TX PUSCH DMRS data",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_POS": {
|
||||
"file_name_prefix": "raw-ce-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Frequency-domain raw channel estimates (at DMRS positions)",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_FD_CHAN_EST_DMRS_INTERPL": {
|
||||
"file_name_prefix": "raw-inter-ce-fd-data",
|
||||
"scope": "gNB",
|
||||
"description": "Interpolcated Frequency-domain raw channel estimates",
|
||||
"serialization_scheme": ["subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"GNB_PHY_UL_PAYLOAD_RX_BITS": {
|
||||
"file_name_prefix": "rx-payload-bits",
|
||||
"scope": "gNB",
|
||||
"description": "Received PUSCH payload bits",
|
||||
"serialization_scheme": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
# UE messages
|
||||
"UE_PHY_UL_SCRAMBLED_TX_BITS": {
|
||||
"file_name_prefix": "tx-scrambled-bits",
|
||||
"scope": "UE",
|
||||
"description": "Transmitted scrambled PUSCH bits",
|
||||
"serialization_scheme": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
"UE_PHY_UL_PAYLOAD_TX_BITS": {
|
||||
"file_name_prefix": "tx-payload-bits",
|
||||
"scope": "UE",
|
||||
"description": "Transmitted PUSCH payload bits",
|
||||
"serialization_scheme": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
},
|
||||
}
|
||||
|
||||
def read_and_store_tracer_messages(shm_reading, bufIdx, num_messages, config_meta_data,
|
||||
global_info, update_timestamp=False):
|
||||
"""
|
||||
Read tracer messages from shared memory while they belong to the same frame and slot.
|
||||
Continuously reads messages until a different frame/slot is encountered.
|
||||
|
||||
Args:
|
||||
shm_reading: Shared memory reading handle
|
||||
bufIdx: Current buffer index
|
||||
num_messages: Maximum number of messages to read (used as safeguard)
|
||||
config_meta_data: Configuration metadata
|
||||
global_info: Global information dict (for timestamp on first message)
|
||||
update_timestamp: If True, update global_info timestamp from first message
|
||||
|
||||
Returns:
|
||||
tuple: (updated_bufIdx, collected_metafiles, last_captured_data)
|
||||
"""
|
||||
tracer_msgs_identities = config_meta_data["data_recording_config"]["tracer_msgs_identities"]
|
||||
collected_metafiles = []
|
||||
captured_data = None
|
||||
|
||||
# Get sync header to check frame and slot
|
||||
sync_header_msg, sync_header_msg_length = data_recording_messages_def.get_sync_header_msg_list()
|
||||
|
||||
# Read first message to establish the reference frame and slot
|
||||
timeout = 10 # 10 seconds timeout
|
||||
is_data_in_memory = shm_interface.is_data_available_in_memory(
|
||||
shm_reading, bufIdx, sync_header_msg_length, timeout)
|
||||
|
||||
if not is_data_in_memory:
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print("Warning: No data available in shared memory")
|
||||
return bufIdx, collected_metafiles, global_info
|
||||
|
||||
# Get reference frame and slot from first message
|
||||
ref_frame, ref_slot = shm_interface.get_frame_slot_start(
|
||||
shm_reading, bufIdx, sync_header_msg, sync_header_msg_length)
|
||||
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"\nRecord number: {global_info['record_idx']}, Reference Frame: {ref_frame}, Slot: {ref_slot}")
|
||||
|
||||
idx = 0
|
||||
while idx < num_messages:
|
||||
time.sleep(0.0015)
|
||||
|
||||
# Check if data is available
|
||||
is_data_in_memory = shm_interface.is_data_available_in_memory(
|
||||
shm_reading, bufIdx, sync_header_msg_length, timeout)
|
||||
|
||||
if not is_data_in_memory:
|
||||
break
|
||||
|
||||
# Get frame and slot of current message before reading
|
||||
current_frame, current_slot = shm_interface.get_frame_slot_start(
|
||||
shm_reading, bufIdx, sync_header_msg, sync_header_msg_length)
|
||||
|
||||
# Check if we're still on the same frame and slot
|
||||
if current_frame != ref_frame or current_slot != ref_slot:
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"Frame/Slot changed: ({ref_frame}, {ref_slot}) -> ({current_frame}, {current_slot}). Stopping at message {idx}")
|
||||
# Keep buffer index pointing to this new frame/slot message (don't increment)
|
||||
break
|
||||
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"Reading MSG {idx}: Frame={current_frame}, Slot={current_slot}")
|
||||
|
||||
# Read the message
|
||||
captured_data, bufIdx = shm_interface.read_data_from_shm(
|
||||
shm_reading, bufIdx, tracer_msgs_identities)
|
||||
|
||||
# Update timestamp from the first message if requested
|
||||
if idx == 0 and update_timestamp:
|
||||
time_stamp, time_stamp_file_name = (
|
||||
sigmf_interface.time_stamp_formating(
|
||||
captured_data["unix_capture_ts_sec"],
|
||||
captured_data["unix_capture_ts_nsec"],
|
||||
global_info["datetime_offset"]))
|
||||
global_info["collection_file"] = (
|
||||
f"{global_info['collection_file_prefix']}-rec-{global_info['record_idx']}-{time_stamp_file_name}")
|
||||
global_info["timestamp"] = time_stamp
|
||||
global_info["frame"] = captured_data["frame"]
|
||||
global_info["slot"]= captured_data["slot"]
|
||||
|
||||
# Write data into files with the given format
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
collected_metafiles.append(sigmf_interface.write_recorded_data_to_sigmf(
|
||||
captured_data, config_meta_data, global_info, global_info['record_idx']))
|
||||
|
||||
idx += 1
|
||||
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f"Total messages read: {idx} for Frame {ref_frame}, Slot {ref_slot}")
|
||||
|
||||
return bufIdx, collected_metafiles, global_info
|
||||
|
||||
def find_sync_buffer_index(shm_reading, sync_info, tracer_msgs_identities, timeout_sec=None):
|
||||
"""
|
||||
Find buffer index that matches sync frame and slot.
|
||||
|
||||
Args:
|
||||
shm_reading: Shared memory reading handle
|
||||
sync_info: Dictionary with 'frame' and 'slot' keys
|
||||
tracer_msgs_identities: Tracer message identities
|
||||
timeout_sec: Timeout in seconds (None for no timeout)
|
||||
|
||||
Returns:
|
||||
bufIdx: Buffer index where sync match was found (points to the matching message)
|
||||
|
||||
Raises:
|
||||
Exception if timeout occurs before sync is found
|
||||
"""
|
||||
bufIdx = 0
|
||||
timeout_sync = time.time() + timeout_sec if timeout_sec else None
|
||||
|
||||
while True:
|
||||
time.sleep(0.0035) # 2.3 ms = latency of T tracer to capture data from the RAN
|
||||
saved_bufIdx = bufIdx # Save index BEFORE read
|
||||
captured_data, bufIdx = shm_interface.read_data_from_shm(
|
||||
shm_reading, bufIdx, tracer_msgs_identities)
|
||||
print(f" [NR Data Sync] NR Data (Frame:{captured_data['frame']}, Slot:{captured_data['slot']}) | "
|
||||
f"Target (Frame:{sync_info['frame']}, Slot:{sync_info['slot']})")
|
||||
|
||||
if (captured_data["frame"] == sync_info["frame"] and
|
||||
captured_data["slot"] == sync_info["slot"]):
|
||||
print(" [NR Data Sync] Sync pass")
|
||||
return saved_bufIdx # Return index pointing to the matching message
|
||||
|
||||
if timeout_sync and time.time() > timeout_sync:
|
||||
raise Exception(
|
||||
"ERROR: Data Recording NO Sync Found, check Tracer Services if they are connected!")
|
||||
|
||||
def send_5g_nr_msgs_request(barrier, shm_writing, shm_reading, args):
|
||||
# send Tracer Control Message request to T-Tracers Apps
|
||||
# -------------------------------------------
|
||||
# It consists of the following fields:
|
||||
# Record action
|
||||
# Number of requested Tracer Messages
|
||||
# Requested Tracer Messages ID 1, …, ID N
|
||||
# Number of records to be recorded in slots
|
||||
# Start SFN: Frame Index to start data collection from it, but not yet used
|
||||
shm_interface.write_shm_task(barrier,shm_writing, args)
|
||||
|
||||
def generic_data_conversion_service(node_shm, config_meta_data, tracer_nodes, sync_info=None):
|
||||
"""Generic data conversion service for N tracer nodes.
|
||||
|
||||
Args:
|
||||
node_shm: dict {node_id: {"reading": shm, "writing": shm}}
|
||||
config_meta_data: Config metadata dict
|
||||
tracer_nodes: dict {node_id: args} for nodes with tracer messages
|
||||
sync_info: Optional sync info dict with 'frame' and 'slot'. If provided,
|
||||
each node's buffer index is synced to this frame/slot.
|
||||
"""
|
||||
record_idx = 0
|
||||
prev_frame = -1
|
||||
prev_slot = -1
|
||||
|
||||
tracer_msgs_identities = config_meta_data["data_recording_config"]["tracer_msgs_identities"]
|
||||
global_info = config_meta_data["data_recording_config"]["global_info"]
|
||||
|
||||
# Find sync buffer indices for each node if sync_info provided
|
||||
node_buf_idx = {}
|
||||
for node_id in tracer_nodes:
|
||||
if sync_info:
|
||||
print(f" [NR Data Sync] Finding memory index for node {node_id}")
|
||||
node_buf_idx[node_id] = find_sync_buffer_index(
|
||||
node_shm[node_id]["reading"], sync_info, tracer_msgs_identities, timeout_sec=5)
|
||||
else:
|
||||
node_buf_idx[node_id] = 0
|
||||
|
||||
# Determine num_records from first tracer node
|
||||
num_records = next(iter(tracer_nodes.values())).num_records
|
||||
|
||||
print("\n--- Data Conversion: Reading data from T-tracer Applications ---")
|
||||
for node_id, args in tracer_nodes.items():
|
||||
print(f" [Data Conversion] {node_id}: Requested tracer messages: {args.num_requested_tracer_msgs}")
|
||||
|
||||
# Read data from all tracer nodes
|
||||
while True:
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print("\nRecord number: ", record_idx)
|
||||
if DEBUG_BUFFER_READING:
|
||||
for nid in tracer_nodes:
|
||||
print(f" Buffer Index {nid}: {node_buf_idx[nid]}")
|
||||
time.sleep(0.0035)
|
||||
|
||||
global_info["record_idx"] = record_idx
|
||||
all_metafiles = []
|
||||
|
||||
for i, (node_id, args) in enumerate(tracer_nodes.items()):
|
||||
buf_idx, metafiles, global_info = read_and_store_tracer_messages(
|
||||
node_shm[node_id]["reading"], node_buf_idx[node_id],
|
||||
args.num_requested_tracer_msgs, config_meta_data, global_info,
|
||||
update_timestamp=(i == 0))
|
||||
node_buf_idx[node_id] = buf_idx
|
||||
all_metafiles.extend(metafiles)
|
||||
|
||||
# Generate SigMF collection file
|
||||
if config_meta_data["data_recording_config"]["enable_saving_tracer_messages_sigmf"]:
|
||||
data_storage_path = config_meta_data["data_recording_config"]["data_storage_path"]
|
||||
description = global_info["description"]
|
||||
sigmf_interface.save_sigmf_collection(
|
||||
all_metafiles, global_info, description, data_storage_path)
|
||||
|
||||
frame = global_info["frame"]
|
||||
slot = global_info["slot"]
|
||||
|
||||
if frame != prev_frame or slot != prev_slot:
|
||||
record_idx += 1
|
||||
if record_idx >= num_records:
|
||||
break
|
||||
prev_frame = frame
|
||||
prev_slot = slot
|
||||
|
||||
# -------------------------------------------
|
||||
# Cleanup — ensures shared memory and resources are released even on Ctrl+C
|
||||
# -------------------------------------------
|
||||
# Module-level dict populated at runtime with resources that need cleanup.
|
||||
# Each key is optional; the cleanup function checks before acting.
|
||||
_cleanup_state = {"done": False}
|
||||
|
||||
|
||||
def cleanup_resources():
|
||||
"""Release T-tracer shared memory resources."""
|
||||
if _cleanup_state["done"]:
|
||||
return
|
||||
_cleanup_state["done"] = True
|
||||
print("\nCleaning up resources...")
|
||||
|
||||
tracer_nodes = _cleanup_state.get("tracer_nodes", {})
|
||||
node_shm = _cleanup_state.get("node_shm", {})
|
||||
for node_id in tracer_nodes:
|
||||
try:
|
||||
tracer_nodes[node_id].action = "quit"
|
||||
shm_interface.write_shm(node_shm[node_id]["writing"], tracer_nodes[node_id])
|
||||
time.sleep(0.5)
|
||||
shm_interface.detach_shm(node_shm[node_id]["reading"])
|
||||
shm_interface.detach_shm(node_shm[node_id]["writing"])
|
||||
shm_interface.remove_shm(node_shm[node_id]["reading"])
|
||||
shm_interface.remove_shm(node_shm[node_id]["writing"])
|
||||
except Exception as e:
|
||||
print(f"Warning: {node_id} shared memory cleanup error: {e}")
|
||||
|
||||
|
||||
def register_cleanup():
|
||||
"""Register cleanup_resources with atexit and signal handlers."""
|
||||
atexit.register(cleanup_resources)
|
||||
|
||||
def _signal_handler(signum, frame):
|
||||
print(f"\nSignal {signum} received, shutting down...")
|
||||
cleanup_resources()
|
||||
sys.exit(130)
|
||||
|
||||
signal.signal(signal.SIGINT, _signal_handler)
|
||||
signal.signal(signal.SIGTERM, _signal_handler)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# -------------------------------------------
|
||||
# Data Control Service
|
||||
## -------------------------------------------
|
||||
# ------------- Configuration --------------
|
||||
# ------------------------------------------
|
||||
# Data Recording Configuration
|
||||
data_recording_config_file = "config/config_data_recording.json"
|
||||
# -------------------------------------------
|
||||
# Configuration
|
||||
# -------------------------------------------
|
||||
# Read and parse the JSON file
|
||||
with open(data_recording_config_file, "r") as file:
|
||||
config_meta_data = json.load(file)
|
||||
# -------------------------------------------
|
||||
# Generic node-based configuration
|
||||
# -------------------------------------------
|
||||
config_meta_data, all_node_args = config_interface.get_data_recording_config(config_meta_data)
|
||||
_nodes_cfg = config_meta_data["data_recording_config"]["nodes"]
|
||||
|
||||
# Build tracer_nodes: only nodes with requested tracer messages
|
||||
tracer_nodes = {nid: args for nid, args in all_node_args.items()
|
||||
if args.requested_tracer_messages}
|
||||
|
||||
# Validate: at least one node must have tracer messages
|
||||
any_tracer = bool(tracer_nodes)
|
||||
if not any_tracer:
|
||||
raise Exception("ERROR: No requested tracer messages IDs are provided")
|
||||
|
||||
# -------------------------------------------
|
||||
# Attach shared memory for each tracer node
|
||||
# -------------------------------------------
|
||||
node_shm = {} # {node_id: {"reading": shm, "writing": shm}}
|
||||
for node_id, args in tracer_nodes.items():
|
||||
node_cfg = next(n for n in _nodes_cfg if n["id"] == node_id)
|
||||
shm_cfg = node_cfg.get("shared_mem_config", {})
|
||||
read_path = shm_cfg.get("read_path")
|
||||
write_path = shm_cfg.get("write_path")
|
||||
proj_id = shm_cfg.get("project_id")
|
||||
if not read_path or not write_path or proj_id is None:
|
||||
raise Exception(
|
||||
f"ERROR: Node '{node_id}' has tracer messages but missing 'shared_mem_config' "
|
||||
f"(read_path, write_path, project_id) in config.")
|
||||
shm_writing = shm_interface.attach_shm(write_path, proj_id)
|
||||
shm_reading = shm_interface.attach_shm(read_path, proj_id)
|
||||
node_shm[node_id] = {"reading": shm_reading, "writing": shm_writing}
|
||||
|
||||
# get general message header list
|
||||
general_msg_header_list, general_message_header_length = \
|
||||
data_recording_messages_def.get_general_msg_header_list()
|
||||
|
||||
# Add supported OAI Tracer Messages
|
||||
config_meta_data["data_recording_config"]["supported_oai_tracer_messages"] = supported_oai_tracer_messages
|
||||
# Add global info
|
||||
config_meta_data["data_recording_config"]["global_info"] = global_info
|
||||
|
||||
# -------------------------------------------
|
||||
# Initialization
|
||||
# -------------------------------------------
|
||||
# send Tracer Control Message request to T-Tracers Apps
|
||||
# It consists of the following fields:
|
||||
# Config action
|
||||
# IP address length
|
||||
# IP address
|
||||
# Port Number
|
||||
# Configure T-Tracers via shared memory for each tracer node
|
||||
for node_id, args in tracer_nodes.items():
|
||||
args.action = "config"
|
||||
shm_interface.write_shm(node_shm[node_id]["writing"], args)
|
||||
|
||||
time.sleep(0.5) # wait for the config to be applied
|
||||
|
||||
# -------------------------------------------
|
||||
# Execution
|
||||
# -------------------------------------------
|
||||
# send Tracer Control Message request to T-Tracers Apps
|
||||
# -------------------------------------------
|
||||
# It consists of the following fields:
|
||||
# Record action
|
||||
# Number of requested Tracer Messages
|
||||
# Requested Tracer Messages ID 1, …, ID N
|
||||
# Number of records to be recorded in slots
|
||||
# Start SFN: Frame Index to start data collection from it, useful for future
|
||||
# data sync between gNB and UE but not yet used
|
||||
# -------------------------------------------
|
||||
# -------- Data Collection Service ----------
|
||||
# -------------------------------------------
|
||||
print("Args:")
|
||||
for node_id, args in tracer_nodes.items():
|
||||
args.action = "record"
|
||||
print(f" {node_id} args: ", args)
|
||||
|
||||
# Pre-create thread pool if user would like to create a for loop for N iterations
|
||||
# Outside the loop to avoid initialization overhead.
|
||||
# max_workers and barrier_parties are kept identical: every thread that
|
||||
# enters the pool must also call barrier.wait(), so pool size == barrier size.
|
||||
total_thread_count = len(tracer_nodes)
|
||||
thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=total_thread_count)
|
||||
barrier = threading.Barrier(total_thread_count) if total_thread_count > 0 else None
|
||||
|
||||
# Populate cleanup state and register signal handlers
|
||||
_cleanup_state.update({
|
||||
"tracer_nodes": tracer_nodes,
|
||||
"node_shm": node_shm,
|
||||
})
|
||||
register_cleanup()
|
||||
|
||||
# Start data recording iterations
|
||||
start_time = time.time()
|
||||
print("Send data logging request us:", datetime.now().strftime("%Y%m%d-%H%M%S%f"))
|
||||
|
||||
# -------------------------------------------
|
||||
# Submit NR tracer threads for all tracer nodes
|
||||
# -------------------------------------------
|
||||
futures = []
|
||||
for node_id, args in tracer_nodes.items():
|
||||
shm = node_shm[node_id]
|
||||
tracer_thread = thread_pool.submit(
|
||||
send_5g_nr_msgs_request, barrier, shm["writing"], shm["reading"], args)
|
||||
futures.append(tracer_thread)
|
||||
|
||||
# Wait for all threads to complete
|
||||
concurrent.futures.wait(futures)
|
||||
|
||||
# -------------------------------------------
|
||||
# Check data availability for each tracer node
|
||||
# -------------------------------------------
|
||||
bufIdx = 0
|
||||
timeout = 10 # 10 seconds from now
|
||||
|
||||
for node_id in tracer_nodes:
|
||||
is_data_in_memory = shm_interface.is_data_available_in_memory(
|
||||
node_shm[node_id]["reading"], bufIdx, general_message_header_length, timeout)
|
||||
|
||||
if not is_data_in_memory:
|
||||
print(f"Error: {node_id}: Check T-Tracer APP, check IPs and Ports")
|
||||
print(f"Error: {node_id}: If IPs and Ports are correct, re-run the hanging app. "
|
||||
"It seems the socket was not closed properly")
|
||||
raise Exception(
|
||||
f"ERROR: Time out, check if {node_id} T-Tracer APP connected to stack")
|
||||
|
||||
# -------------------------------------------
|
||||
# Sync and convert data
|
||||
# -------------------------------------------
|
||||
common_utils.write_config_data_recording_app_json(config_meta_data)
|
||||
sync_info = {}
|
||||
|
||||
# Step 1: Sync NR tracer nodes (find latest common frame/slot across all NR sources)
|
||||
if len(tracer_nodes) > 1:
|
||||
node_shm_readings = {nid: node_shm[nid]["reading"] for nid in tracer_nodes}
|
||||
sync_info = sync_service.sync_multiple_nr_nodes(node_shm_readings)
|
||||
print(f"\n--- NR Node Sync: Sync point: frame={sync_info['frame']}, slot={sync_info['slot']} ---")
|
||||
|
||||
# Step 2: Read and store NR tracer data
|
||||
if tracer_nodes:
|
||||
do_sync = bool(sync_info)
|
||||
generic_data_conversion_service(
|
||||
node_shm, config_meta_data, tracer_nodes, sync_info if do_sync else None)
|
||||
|
||||
# measure Elapsed time
|
||||
time_elapsed = time.time() - start_time
|
||||
time_elapsed_ms = int(time_elapsed * 1000)
|
||||
print(
|
||||
"Elapsed time of getting Requested Messages and writing data and meta data files:",
|
||||
colored(time_elapsed_ms, "yellow"),
|
||||
"ms",)
|
||||
|
||||
# Shutdown thread pool after all iterations complete
|
||||
thread_pool.shutdown(wait=True)
|
||||
|
||||
# Run cleanup (also registered with atexit and signal handlers for abnormal exits)
|
||||
cleanup_resources()
|
||||
|
||||
print("End of the RF Data Recording API")
|
||||
@@ -1,12 +1,11 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief Data Recording common utilities
|
||||
# brief Common utilities of Data Recording App
|
||||
|
||||
import os
|
||||
import json
|
||||
|
||||
|
||||
def real_to_complex(real_vector):
|
||||
# Ensure the length of the real vector is even
|
||||
if len(real_vector) % 2 != 0:
|
||||
@@ -34,7 +33,7 @@ def write_config_data_recording_app_json(config_meta_data):
|
||||
# Specify the file name
|
||||
output_file = (
|
||||
config_meta_data["data_recording_config"]["data_storage_path"]
|
||||
+ "config_data_recording_app.json"
|
||||
+ "config_data_recording_app_extended.json"
|
||||
)
|
||||
# Ensure the directory exists
|
||||
os.makedirs(os.path.dirname(output_file), exist_ok=True)
|
||||
@@ -43,6 +42,6 @@ def write_config_data_recording_app_json(config_meta_data):
|
||||
with open(output_file, "w") as file:
|
||||
try:
|
||||
json.dump(config_meta_data, file, indent=4)
|
||||
print(f"JSON file created successfully at {output_file}")
|
||||
print(f" [Config] JSON file saved: {output_file}")
|
||||
except Exception as e:
|
||||
print(f"Failed to create JSON file: {e}")
|
||||
print(f" [Config] Failed to create JSON file: {e}")
|
||||
|
||||
@@ -1,18 +1,17 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief Data Recording App Configuration Interface
|
||||
# brief Configuration Interface of Data Recording App
|
||||
|
||||
import yaml
|
||||
import json
|
||||
import argparse
|
||||
|
||||
|
||||
# Function to read the main configuration file in JSON format
|
||||
def read_main_config_file_json(main_config_file: str) -> dict:
|
||||
"""
|
||||
Reads main config file.
|
||||
"""
|
||||
|
||||
# read general parameter set from yaml config file
|
||||
with open(main_config_file, "r") as file:
|
||||
main_config = json.load(file)
|
||||
@@ -20,22 +19,11 @@ def read_main_config_file_json(main_config_file: str) -> dict:
|
||||
return main_config
|
||||
|
||||
|
||||
# Function to read the main configuration file in YAML format
|
||||
def read_main_config_file(main_config_file: str) -> dict:
|
||||
"""
|
||||
Reads main config file.
|
||||
"""
|
||||
# read general parameter set from yaml config file
|
||||
with open(main_config_file, "r") as file:
|
||||
main_config = yaml.load(file, Loader=yaml.Loader)
|
||||
|
||||
return main_config
|
||||
|
||||
|
||||
# Function to parse the OAI T_messages file and get the index of a given string
|
||||
def parse_message_file(file_path):
|
||||
with open(file_path, "r") as file:
|
||||
content = file.readlines()
|
||||
|
||||
# Extract lines that start with 'ID' and remove the 'ID = ' prefix
|
||||
tracer_msgs_identities = [
|
||||
line.strip().replace("ID = ", "")
|
||||
@@ -65,12 +53,18 @@ def get_requested_tracer_msgs_indices(requested_tracer_messages, tracer_msgs_ide
|
||||
return req_tracer_msgs_indices
|
||||
|
||||
|
||||
# Function to get the data recording configuration
|
||||
def get_data_recording_config(config_meta_data):
|
||||
parser = argparse.ArgumentParser(description="request messages IDs")
|
||||
ue_args = parser.parse_args()
|
||||
gnb_args = parser.parse_args()
|
||||
# Helper: return the first node in the nodes list whose type matches the given type string
|
||||
def get_node_by_type(nodes, node_type):
|
||||
for node in nodes:
|
||||
if node.get("type") == node_type:
|
||||
return node
|
||||
return None
|
||||
|
||||
|
||||
# Function to get the data recording configuration.
|
||||
# Returns (config_meta_data, node_args_dict) where node_args_dict is
|
||||
# { node_id: Namespace } with one entry per node defined in the config.
|
||||
def get_data_recording_config(config_meta_data):
|
||||
# get Tracer Messages IDs from the T-Tracer Messages file
|
||||
# MSG list order should be similar to the T-Tracer Messages txt file
|
||||
# Be sure that you are using the same T_messages file that is used in the OAI project
|
||||
@@ -81,44 +75,33 @@ def get_data_recording_config(config_meta_data):
|
||||
config_meta_data["data_recording_config"]["t_tracer_message_definition_file"])
|
||||
config_meta_data["data_recording_config"]["tracer_msgs_identities"] = tracer_msgs_identities
|
||||
|
||||
# get requested tracer messages indices for gNB
|
||||
gnb_args.requested_tracer_messages = \
|
||||
config_meta_data["data_recording_config"]["base_station"]["requested_tracer_messages"]
|
||||
# get requested tracer messages indices for UE
|
||||
ue_args.requested_tracer_messages = \
|
||||
config_meta_data["data_recording_config"]["user_equipment"]["requested_tracer_messages"]
|
||||
nodes = config_meta_data["data_recording_config"]["nodes"]
|
||||
|
||||
# check if gnb_requested_tracer_messages is not empty
|
||||
if gnb_args.requested_tracer_messages:
|
||||
config_meta_data["data_recording_config"]["base_station"][
|
||||
"req_tracer_msgs_indices"] = get_requested_tracer_msgs_indices(
|
||||
gnb_args.requested_tracer_messages, tracer_msgs_identities)
|
||||
# Build per-node args generically
|
||||
node_args_dict = {}
|
||||
for node in nodes:
|
||||
parser = argparse.ArgumentParser(description="request messages IDs")
|
||||
node_args = parser.parse_args()
|
||||
|
||||
# get gNB Trace Messages
|
||||
gnb_args.num_records = config_meta_data["data_recording_config"]["num_records"]
|
||||
gnb_args.start_frame_number = config_meta_data["data_recording_config"]["start_frame_number"]
|
||||
gnb_args.req_tracer_msgs_indices = \
|
||||
config_meta_data["data_recording_config"]["base_station"]["req_tracer_msgs_indices"]
|
||||
gnb_args.num_requested_tracer_msgs = len(gnb_args.req_tracer_msgs_indices)
|
||||
# Split the string into IP and port
|
||||
gnb_args.IPaddress, gnb_args.port = config_meta_data["data_recording_config"][
|
||||
"tracer_service_baseStation_address"].split(":")
|
||||
gnb_args.bytes_IPaddress = bytes(gnb_args.IPaddress, "utf-8")
|
||||
node_id = node.get("id")
|
||||
node_args.node_id = node_id
|
||||
node_args.type = node.get("type")
|
||||
node_args.requested_tracer_messages = node.get("requested_tracer_messages", [])
|
||||
|
||||
# check if ue_requested_tracer_messages is not empty
|
||||
if ue_args.requested_tracer_messages:
|
||||
config_meta_data["data_recording_config"]["user_equipment"][
|
||||
"req_tracer_msgs_indices"] = get_requested_tracer_msgs_indices(
|
||||
ue_args.requested_tracer_messages, tracer_msgs_identities)
|
||||
if node_args.requested_tracer_messages:
|
||||
node["req_tracer_msgs_indices"] = get_requested_tracer_msgs_indices(
|
||||
node_args.requested_tracer_messages, tracer_msgs_identities)
|
||||
|
||||
# get UE Trace Messages
|
||||
ue_args.num_records = config_meta_data["data_recording_config"]["num_records"]
|
||||
ue_args.start_frame_number = config_meta_data["data_recording_config"]["start_frame_number"]
|
||||
ue_args.req_tracer_msgs_indices = \
|
||||
config_meta_data["data_recording_config"]["user_equipment"]["req_tracer_msgs_indices"]
|
||||
ue_args.num_requested_tracer_msgs = len(ue_args.req_tracer_msgs_indices)
|
||||
ue_args.IPaddress, ue_args.port = config_meta_data["data_recording_config"][
|
||||
"tracer_service_userEquipment_address"].split(":")
|
||||
ue_args.bytes_IPaddress = bytes(ue_args.IPaddress, "utf-8")
|
||||
node_args.num_records = config_meta_data["data_recording_config"]["num_records"]
|
||||
node_args.start_frame_number = config_meta_data["data_recording_config"]["start_frame_number"]
|
||||
node_args.req_tracer_msgs_indices = node["req_tracer_msgs_indices"]
|
||||
node_args.num_requested_tracer_msgs = len(node_args.req_tracer_msgs_indices)
|
||||
# Split the string into IP and port
|
||||
tracer_addr = node.get("tracer_service_address", "")
|
||||
if tracer_addr:
|
||||
node_args.IPaddress, node_args.port = tracer_addr.split(":")
|
||||
node_args.bytes_IPaddress = bytes(node_args.IPaddress, "utf-8")
|
||||
|
||||
return config_meta_data, gnb_args, ue_args
|
||||
node_args_dict[node_id] = node_args
|
||||
|
||||
return config_meta_data, node_args_dict
|
||||
|
||||
@@ -3,6 +3,32 @@
|
||||
# ---------------------------------------------------------------------
|
||||
# brief defination of captured data recording messages
|
||||
|
||||
# Get Common Sync Header - number of bytes
|
||||
def get_sync_header_msg_list():
|
||||
"""
|
||||
shared memory layout written from the app:
|
||||
=================================
|
||||
msg_id (uint8) message type ID
|
||||
frame (uint16)
|
||||
slot (uint8)
|
||||
unix_capture_ts_sec (uint32) Unix epoch seconds
|
||||
unix_capture_ts_nsec (uint32) nanoseconds [0, 999999999]
|
||||
|
||||
"""
|
||||
# Get Common Sync Header - number of bytes
|
||||
sync_header_msg = {
|
||||
"msg_id": 2,
|
||||
"frame": 2,
|
||||
"slot": 1,
|
||||
"unix_capture_ts_sec": 4,
|
||||
"unix_capture_ts_nsec": 4,
|
||||
}
|
||||
# initial number of bytes to read to get data
|
||||
sync_header_msg_length = 0
|
||||
for key, value in sync_header_msg.items():
|
||||
sync_header_msg_length = sync_header_msg_length + value
|
||||
return sync_header_msg, sync_header_msg_length
|
||||
|
||||
# Data Collection Trace Messages - General message structure - number of bytes
|
||||
def get_general_msg_header_list():
|
||||
"""
|
||||
@@ -11,8 +37,8 @@ def get_general_msg_header_list():
|
||||
msg_id (uint8) message type ID
|
||||
frame (uint16)
|
||||
slot (uint8)
|
||||
datetime_yyyymmdd (uint32)
|
||||
datetime_hhmmssmmm (uint32)
|
||||
unix_capture_ts_sec (uint32) Unix epoch seconds
|
||||
unix_capture_ts_nsec (uint32) nanoseconds [0, 999999999]
|
||||
frame_type (uint8)
|
||||
freq_range (uint8)
|
||||
subcarrier_spacing (uint8)
|
||||
@@ -46,8 +72,8 @@ def get_general_msg_header_list():
|
||||
"msg_id": 2,
|
||||
"frame": 2,
|
||||
"slot": 1,
|
||||
"datetime_yyyymmdd": 4,
|
||||
"datetime_hhmmssmmm": 4,
|
||||
"unix_capture_ts_sec": 4,
|
||||
"unix_capture_ts_nsec": 4,
|
||||
"frame_type": 1,
|
||||
"freq_range": 1,
|
||||
"subcarrier_spacing": 1,
|
||||
|
||||
348
common/utils/data_recording/lib/shared_memory_interface.py
Normal file
348
common/utils/data_recording/lib/shared_memory_interface.py
Normal file
@@ -0,0 +1,348 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief Shared Memory Interface for Data Recording App
|
||||
|
||||
import sysv_ipc as ipc
|
||||
import time
|
||||
import struct
|
||||
from lib import data_recording_messages_def
|
||||
import numpy as np
|
||||
from lib import common_utils
|
||||
|
||||
DEBUG_WIRELESS_RECORDED_DATA = True
|
||||
DEBUG_BUFFER_READING = False
|
||||
|
||||
# initialize shared memory
|
||||
def attach_shm(shm_path, project_id):
|
||||
key = ipc.ftok(shm_path, project_id)
|
||||
shm = ipc.SharedMemory(key, 0, 0)
|
||||
shm.attach(0, 0)
|
||||
return shm
|
||||
|
||||
|
||||
def detach_shm(shm):
|
||||
try:
|
||||
shm.detach()
|
||||
print("Shared memory detached successfully.")
|
||||
except ipc.ExistentialError:
|
||||
print("Shared memory segment does not exist.")
|
||||
|
||||
def remove_shm(shm):
|
||||
try:
|
||||
shm.remove()
|
||||
print("Shared memory removed successfully.")
|
||||
except ipc.ExistentialError:
|
||||
print("Shared memory segment does not exist.")
|
||||
|
||||
|
||||
# Write Tracer Control Message
|
||||
def write_shm(shm, args):
|
||||
# Note: Big Endian >, Little Endian <
|
||||
# Note: unsigned char B, signed char b, short h, int I, long long q
|
||||
# Note: float f, double d, string s, char c, bool ?
|
||||
# 1: config
|
||||
# 2: record
|
||||
# 3: quit
|
||||
print(f" [Shared Memory] Write action: {args.action}")
|
||||
if args.action == "config":
|
||||
# Determine the length of the IP address
|
||||
ip_length = len(args.bytes_IPaddress) + 1 # String terminator
|
||||
# Construct the format string dynamically
|
||||
format_string = f"{ip_length}s"
|
||||
shm.write(
|
||||
# Config action
|
||||
struct.pack("<B", 1) +
|
||||
struct.pack("<B", ip_length) +
|
||||
struct.pack(format_string, args.bytes_IPaddress) +
|
||||
struct.pack("<h", int(args.port))
|
||||
)
|
||||
print(f" [T-Tracer] Config: IP={args.bytes_IPaddress}, port={args.port}")
|
||||
elif args.action == "record":
|
||||
shm.write(
|
||||
# Record action
|
||||
struct.pack("<B", int(2)) +
|
||||
struct.pack("<B", args.num_requested_tracer_msgs) +
|
||||
struct.pack("<{}h".format(len(args.req_tracer_msgs_indices)),
|
||||
*args.req_tracer_msgs_indices,) +
|
||||
struct.pack("<I", args.num_records) +
|
||||
struct.pack("<h", args.start_frame_number)
|
||||
)
|
||||
print(f" [T-Tracer] Record: num_msgs={args.num_requested_tracer_msgs}, "
|
||||
f"msg_ids={args.req_tracer_msgs_indices}, "
|
||||
f"num_records={args.num_records}, "
|
||||
f"start_frame={args.start_frame_number}")
|
||||
elif args.action == "quit":
|
||||
shm.write(struct.pack("<B", int(3))) # Quit action
|
||||
print(" [T-Tracer] Quit")
|
||||
else:
|
||||
print("Unknown action for data recording system!")
|
||||
|
||||
|
||||
# write shared memory task
|
||||
def write_shm_task(barrier, shm_id, args):
|
||||
# Wait for threads to be ready
|
||||
barrier.wait()
|
||||
# send request to related T-Tracer Application
|
||||
write_shm(shm_id, args)
|
||||
|
||||
# check data if avalible in the shared memory
|
||||
def is_data_available_in_memory(shm, bufIdx, general_message_header_length, timeout=10):
|
||||
start_time = time.time()
|
||||
while True:
|
||||
buf = shm.read(bufIdx + general_message_header_length)
|
||||
# Only check bytes in the range [bufIdx, bufIdx + header_length),
|
||||
# not the entire buffer from 0 — earlier records would make sum > 0
|
||||
n_bytes = sum(buf[bufIdx:bufIdx + general_message_header_length])
|
||||
if n_bytes > 0:
|
||||
if DEBUG_BUFFER_READING:
|
||||
print("Data in memory: ", n_bytes, " bytes")
|
||||
return True
|
||||
if (time.time() - start_time) > timeout:
|
||||
break
|
||||
print("Data Recording App: Waiting for Measurements!")
|
||||
time.sleep(1)
|
||||
return False
|
||||
|
||||
# Read data from gNB T-tracer Application
|
||||
def get_frame_slot_start(shm_reading, bufIdx, general_msg_header_list,
|
||||
general_message_header_length):
|
||||
buf = shm_reading.read(bufIdx + general_message_header_length)
|
||||
msg_id = struct.unpack("<H", buf[bufIdx: bufIdx + general_msg_header_list.get("msg_id")])[0]
|
||||
bufIdx += general_msg_header_list.get("msg_id")
|
||||
frame = struct.unpack("<H", buf[bufIdx: bufIdx + general_msg_header_list.get("frame")])[0]
|
||||
bufIdx += general_msg_header_list.get("frame")
|
||||
slot = struct.unpack("B", buf[bufIdx: bufIdx + general_msg_header_list.get("slot")])[0]
|
||||
return frame, slot
|
||||
|
||||
# get MSG header Info
|
||||
def get_msg_header(shm_reading, bufIdx, sync_header_msg, sync_header_length):
|
||||
"""Read the sync header from shared memory and return msg_id, frame, slot, and timestamp.
|
||||
|
||||
Args:
|
||||
shm_reading: Shared memory handle.
|
||||
bufIdx: Current buffer index.
|
||||
sync_header_msg: Dict with field names and byte sizes (from get_sync_header_msg_list).
|
||||
sync_header_length: Total byte length of the sync header.
|
||||
|
||||
Returns:
|
||||
tuple: (msg_id, frame, slot, unix_capture_ts_sec, unix_capture_ts_nsec)
|
||||
"""
|
||||
buf = shm_reading.read(bufIdx + sync_header_length)
|
||||
offset = bufIdx
|
||||
msg_id = struct.unpack('<H', buf[offset:offset + sync_header_msg["msg_id"]])[0]
|
||||
offset += sync_header_msg["msg_id"]
|
||||
frame = struct.unpack('<H', buf[offset:offset + sync_header_msg["frame"]])[0]
|
||||
offset += sync_header_msg["frame"]
|
||||
slot = struct.unpack('B', buf[offset:offset + sync_header_msg["slot"]])[0]
|
||||
offset += sync_header_msg["slot"]
|
||||
unix_capture_ts_sec = struct.unpack('<I', buf[offset:offset + sync_header_msg["unix_capture_ts_sec"]])[0]
|
||||
offset += sync_header_msg["unix_capture_ts_sec"]
|
||||
unix_capture_ts_nsec = struct.unpack('<I', buf[offset:offset + sync_header_msg["unix_capture_ts_nsec"]])[0]
|
||||
return msg_id, frame, slot, unix_capture_ts_sec, unix_capture_ts_nsec
|
||||
|
||||
|
||||
# Get MSG ID from Shared memory
|
||||
def get_msg_id_from_shm(shm_reading, bufIdx, msg_id_length):
|
||||
buf = shm_reading.read(bufIdx + msg_id_length)
|
||||
msg_id = struct.unpack("<H", buf[bufIdx: bufIdx + msg_id_length])[0]
|
||||
return msg_id
|
||||
|
||||
# Read data from Shared memory based Data Conversion Service message structure
|
||||
def read_msg_data_from_shm(shm, bufIdx, tracer_msgs_identities):
|
||||
# read sync header to move buffer index
|
||||
# get general message header list
|
||||
general_msg_header_list, general_message_header_length = \
|
||||
data_recording_messages_def.get_general_msg_header_list()
|
||||
buf = shm.read(bufIdx + general_message_header_length)
|
||||
n_bytes = sum(buf)
|
||||
if n_bytes == 0:
|
||||
raise Exception('ERROR: No data available in memory')
|
||||
msg_id = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("msg_id")])[0]
|
||||
bufIdx += general_msg_header_list.get("msg_id")
|
||||
frame = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("frame")])[0]
|
||||
bufIdx += general_msg_header_list.get("frame")
|
||||
slot = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("slot")])[0]
|
||||
bufIdx += general_msg_header_list.get("slot")
|
||||
# get Unix capture timestamp (sec, nsec) — Unix epoch 1970
|
||||
unix_capture_ts_sec = \
|
||||
struct.unpack('<I', buf[bufIdx:bufIdx+general_msg_header_list.get("unix_capture_ts_sec")])[0]
|
||||
bufIdx += general_msg_header_list.get("unix_capture_ts_sec")
|
||||
unix_capture_ts_nsec = \
|
||||
struct.unpack('<I', buf[bufIdx:bufIdx+general_msg_header_list.get("unix_capture_ts_nsec")])[0]
|
||||
bufIdx += general_msg_header_list.get("unix_capture_ts_nsec")
|
||||
# Derive timestamp in seconds and nanoseconds (already in Unix epoch)
|
||||
timestamp_seconds = unix_capture_ts_sec
|
||||
timestamp_nseconds = unix_capture_ts_nsec
|
||||
# get frame type
|
||||
frame_type = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("frame_type")])[0]
|
||||
bufIdx += general_msg_header_list.get("frame_type")
|
||||
# get frequency range
|
||||
freq_range = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("freq_range")])[0]
|
||||
bufIdx += general_msg_header_list.get("freq_range")
|
||||
# get subcarrier spacing
|
||||
subcarrier_spacing = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("subcarrier_spacing")])[0]
|
||||
bufIdx += general_msg_header_list.get("subcarrier_spacing")
|
||||
# get cyclic prefix
|
||||
cyclic_prefix = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("cyclic_prefix")])[0]
|
||||
bufIdx += general_msg_header_list.get("cyclic_prefix")
|
||||
# get symbols per slot
|
||||
symbols_per_slot = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("symbols_per_slot")])[0]
|
||||
bufIdx += general_msg_header_list.get("symbols_per_slot")
|
||||
# get Nid cell
|
||||
Nid_cell = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("Nid_cell")])[0]
|
||||
bufIdx += general_msg_header_list.get("Nid_cell")
|
||||
# get rnti
|
||||
rnti = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("rnti")])[0]
|
||||
bufIdx += general_msg_header_list.get("rnti")
|
||||
# get rb size
|
||||
rb_size = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("rb_size")])[0]
|
||||
bufIdx += general_msg_header_list.get("rb_size")
|
||||
# get rb start
|
||||
rb_start = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("rb_start")])[0]
|
||||
bufIdx += general_msg_header_list.get("rb_start")
|
||||
# get start symbol index
|
||||
start_symbol_index = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("start_symbol_index")])[0]
|
||||
bufIdx += general_msg_header_list.get("start_symbol_index")
|
||||
# get number of symbols
|
||||
nr_of_symbols = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("nr_of_symbols")])[0]
|
||||
bufIdx += general_msg_header_list.get("nr_of_symbols")
|
||||
# get qam modulation order
|
||||
qam_mod_order = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("qam_mod_order")])[0]
|
||||
bufIdx += general_msg_header_list.get("qam_mod_order")
|
||||
# get mcs index
|
||||
mcs_index = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("mcs_index")])[0]
|
||||
bufIdx += general_msg_header_list.get("mcs_index")
|
||||
# get mcs table
|
||||
mcs_table = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("mcs_table")])[0]
|
||||
bufIdx += general_msg_header_list.get("mcs_table")
|
||||
# get number of layers
|
||||
nrOfLayers = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("nrOfLayers")])[0]
|
||||
bufIdx += general_msg_header_list.get("nrOfLayers")
|
||||
# get transform precoding
|
||||
transform_precoding = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("transform_precoding")])[0]
|
||||
bufIdx += general_msg_header_list.get("transform_precoding")
|
||||
# get dmrs config type
|
||||
dmrs_config_type = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("dmrs_config_type")])[0]
|
||||
bufIdx += general_msg_header_list.get("dmrs_config_type")
|
||||
# get ul dmrs symb pos
|
||||
ul_dmrs_symb_pos = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("ul_dmrs_symb_pos")])[0]
|
||||
bufIdx += general_msg_header_list.get("ul_dmrs_symb_pos")
|
||||
# get number dmrs symbols
|
||||
number_dmrs_symbols = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("number_dmrs_symbols")])[0]
|
||||
bufIdx += general_msg_header_list.get("number_dmrs_symbols")
|
||||
# get dmrs port
|
||||
dmrs_port = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("dmrs_port")])[0]
|
||||
bufIdx += general_msg_header_list.get("dmrs_port")
|
||||
# get dmrs scid
|
||||
dmrs_scid = struct.unpack('<H', buf[bufIdx:bufIdx+general_msg_header_list.get("dmrs_scid")])[0]
|
||||
bufIdx += general_msg_header_list.get("dmrs_scid")
|
||||
# get nb antennas rx for gNB or nb antennas tx for UE
|
||||
nb_antennas = struct.unpack('B', buf[bufIdx:bufIdx+general_msg_header_list.get("nb_antennas")])[0]
|
||||
bufIdx += general_msg_header_list.get("nb_antennas")
|
||||
# get number of bits
|
||||
number_of_bits = struct.unpack('<I', buf[bufIdx:bufIdx+general_msg_header_list.get("number_of_bits")])[0]
|
||||
bufIdx += general_msg_header_list.get("number_of_bits")
|
||||
# get length of bytes
|
||||
length_bytes = struct.unpack('<I', buf[bufIdx:bufIdx+general_msg_header_list.get("length_bytes")])[0]
|
||||
bufIdx += general_msg_header_list.get("length_bytes")
|
||||
# print all captured data
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print("\n" + "="*80)
|
||||
print(f"UL CONFIG METADATA - {tracer_msgs_identities[msg_id]}")
|
||||
print(f"Unix TS: {unix_capture_ts_sec}.{unix_capture_ts_nsec:09d} | MSG ID: {msg_id}")
|
||||
print("-"*80)
|
||||
print(f"Frame: {frame:<5} Slot: {slot:<5} Nid Cell: {Nid_cell:<5} RNTI: {rnti:<5}")
|
||||
print(f"Frame Type: {frame_type} Freq Range: {freq_range} | SCS: {subcarrier_spacing} Hz Cyclic Prefix: {cyclic_prefix} Symbols/Slot: {symbols_per_slot}")
|
||||
print("-"*80)
|
||||
print(f"RB: Start={rb_start:<4} Size={rb_size:<4} | Symbol Allocation: Start={start_symbol_index:<3} Count={nr_of_symbols:<3}")
|
||||
print(f"MCS: Index={mcs_index:<3} Table={mcs_table} | QAM Order: {qam_mod_order} Layers: {nrOfLayers} Transform Precoding: {transform_precoding}")
|
||||
print("-"*80)
|
||||
print(f"DMRS: Config Type={dmrs_config_type} Symb Pos={ul_dmrs_symb_pos} ({number_dmrs_symbols} symbols)")
|
||||
print(f" Port={dmrs_port} SCID={dmrs_scid} | Antennas: {nb_antennas}")
|
||||
print("-"*80)
|
||||
print(f"Data: {number_of_bits} bits ({length_bytes} bytes)")
|
||||
print("="*80)
|
||||
# raise exception if time stamp is zero, it means that the data is not recorded yet
|
||||
if unix_capture_ts_sec == 0 and unix_capture_ts_nsec == 0:
|
||||
raise Exception("ERROR: Time stamp is zero, data is not recorded yet or something wrong, check logs!")
|
||||
# get recorded data
|
||||
buf = shm.read(bufIdx + length_bytes)
|
||||
captured_data = {}
|
||||
# If message is bit message, store data in bytes
|
||||
# then the field number_of_bits should be not zero
|
||||
if "_BITS" in tracer_msgs_identities[msg_id]:
|
||||
recorded_data = struct.unpack("<" + int(length_bytes) * 'B', buf[bufIdx:bufIdx + length_bytes])
|
||||
bufIdx += length_bytes
|
||||
# convert data in bytes to bits
|
||||
bits_vector = []
|
||||
for byte in recorded_data:
|
||||
bits_vector.extend([int(bit) for bit in format(int(byte), '08b')])
|
||||
captured_data["sigmf_data_type"] = "ri8_le"
|
||||
# convert to uint8
|
||||
captured_data["recorded_data"] = np.asarray(bits_vector).astype(np.uint8)
|
||||
else:
|
||||
recorded_data = struct.unpack("<" + int(length_bytes/2) * 'h', buf[bufIdx:bufIdx + length_bytes])
|
||||
bufIdx += length_bytes
|
||||
# Convert real data to complext data
|
||||
# converting list to array
|
||||
recorded_data = np.asarray(recorded_data)
|
||||
recorded_data_complex = common_utils.real_to_complex(recorded_data)
|
||||
captured_data["sigmf_data_type"] = "cf32_le"
|
||||
# convert to complex64
|
||||
captured_data["recorded_data"] = recorded_data_complex.astype(np.complex64)
|
||||
# store data in dictonary
|
||||
captured_data["message_id"] = msg_id
|
||||
captured_data["message_type"] = tracer_msgs_identities[msg_id]
|
||||
captured_data["frame"] = frame
|
||||
captured_data["slot"] = slot
|
||||
captured_data["unix_capture_ts_sec"] = unix_capture_ts_sec
|
||||
captured_data["unix_capture_ts_nsec"] = unix_capture_ts_nsec
|
||||
captured_data["timestamp_seconds"] = timestamp_seconds
|
||||
captured_data["timestamp_nseconds"] = timestamp_nseconds
|
||||
captured_data["frame_type"] = frame_type
|
||||
captured_data["freq_range"] = freq_range
|
||||
captured_data["subcarrier_spacing"] = subcarrier_spacing
|
||||
captured_data["cyclic_prefix"] = cyclic_prefix
|
||||
# captured_data["symbols_per_slot"] = symbols_per_slot ... not used
|
||||
captured_data["Nid_cell"] = Nid_cell
|
||||
captured_data["rnti"] = rnti
|
||||
captured_data["rb_size"] = rb_size
|
||||
captured_data["rb_start"] = rb_start
|
||||
captured_data["start_symbol_index"] = start_symbol_index
|
||||
captured_data["nr_of_symbols"] = nr_of_symbols
|
||||
captured_data["qam_mod_order"] = qam_mod_order
|
||||
captured_data["mcs_index"] = mcs_index
|
||||
captured_data["mcs_table"] = mcs_table
|
||||
captured_data["nrOfLayers"] = nrOfLayers
|
||||
captured_data["transform_precoding"] = transform_precoding
|
||||
captured_data["dmrs_config_type"] = dmrs_config_type
|
||||
captured_data["ul_dmrs_symb_pos"] = ul_dmrs_symb_pos
|
||||
captured_data["number_dmrs_symbols"] = number_dmrs_symbols
|
||||
captured_data["dmrs_port"] = dmrs_port
|
||||
captured_data["dmrs_scid"] = dmrs_scid
|
||||
captured_data["nb_antennas"] = nb_antennas
|
||||
captured_data["number_of_bits"] = number_of_bits
|
||||
return captured_data, bufIdx
|
||||
|
||||
# Read data from Shared memory based Data Conversion Service message structure
|
||||
def read_data_from_shm(shm, bufIdx, tracer_msgs_identities):
|
||||
# print buffer index
|
||||
if DEBUG_BUFFER_READING:
|
||||
print("Buffer Index: ", bufIdx)
|
||||
# Get MSG ID by reading Sync Header
|
||||
sync_header_msg_list, sync_header_message_length = \
|
||||
data_recording_messages_def.get_sync_header_msg_list()
|
||||
msg_id_length = sync_header_msg_list.get("msg_id")
|
||||
|
||||
msg_id = get_msg_id_from_shm(shm, bufIdx, msg_id_length)
|
||||
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(f" [T-Tracer] Reading MSG ID: {msg_id} - {tracer_msgs_identities[msg_id]}")
|
||||
# Get MSG data based on Data Conversion Service message structure
|
||||
# UL IQ/bits messages (default UL data format)
|
||||
|
||||
captured_data, bufIdx = read_msg_data_from_shm(shm, bufIdx, tracer_msgs_identities)
|
||||
|
||||
return captured_data, bufIdx
|
||||
|
||||
@@ -6,11 +6,12 @@
|
||||
import os
|
||||
import sigmf
|
||||
from sigmf import SigMFFile
|
||||
# from sigmf.utils import get_data_type_str
|
||||
import numpy as np
|
||||
from datetime import datetime
|
||||
import yaml
|
||||
from datetime import datetime, timezone
|
||||
from sigmf import SigMFCollection
|
||||
from .wireless_parameters_mapper import map_waveform_metadata_to_wireless_dic, derive_remaining_5gnr_metadata, STANDARDS
|
||||
from .config_interface import get_node_by_type
|
||||
|
||||
DEBUG_WIRELESS_RECORDED_DATA = True
|
||||
|
||||
"""
|
||||
SERIALIZATION_SCHEMES = {
|
||||
@@ -24,21 +25,29 @@ SERIALIZATION_SCHEMES = {
|
||||
"tx-payload-bits": ["bits", "subcarriers", "ofdm_symbols"],
|
||||
}
|
||||
"""
|
||||
STANDARDS = {"5gnr_oai": "5gnr"}
|
||||
|
||||
|
||||
def time_stamp_formating(time_stamp, datetime_offset):
|
||||
# Parse the input string into a datetime object
|
||||
time_stamp_ms_obj = datetime.strptime(time_stamp, "%Y%m%d_%H%M%S%f")
|
||||
def time_stamp_formating(unix_capture_ts_sec, unix_capture_ts_nsec, datetime_offset):
|
||||
# Convert Unix epoch (sec, nsec) to a datetime object (UTC).
|
||||
# Python datetime only supports µs, so we format the nanosecond
|
||||
# fractional part manually to preserve the full 9-digit resolution.
|
||||
dt_obj = datetime.fromtimestamp(unix_capture_ts_sec, tz=timezone.utc)
|
||||
|
||||
# Format the datetime object into the desired output format with milliseconds
|
||||
time_stamp_ms_iso = (
|
||||
time_stamp_ms_obj.strftime("%Y_%m_%dT%H:%M:%S.%f")[:-3] + datetime_offset
|
||||
# Format with nanosecond fractional seconds (9 digits)
|
||||
time_stamp_ns_iso = (
|
||||
dt_obj.strftime("%Y_%m_%dT%H:%M:%S")
|
||||
+ f".{unix_capture_ts_nsec:09d}"
|
||||
+ datetime_offset
|
||||
)
|
||||
time_stamp_ms = time_stamp_ms_iso
|
||||
time_stamp_ms_file_name = time_stamp_ms_iso.replace(":", "_").replace(".", "_")
|
||||
time_stamp_ns_file_name = time_stamp_ns_iso.replace(":", "_").replace(".", "_")
|
||||
|
||||
return time_stamp_ms, time_stamp_ms_file_name
|
||||
return time_stamp_ns_iso, time_stamp_ns_file_name
|
||||
|
||||
|
||||
def _get_node_meta_data(config_meta_data, role):
|
||||
"""Return the meta_data dict for the first node whose type matches the given type string."""
|
||||
node = get_node_by_type(config_meta_data["data_recording_config"]["nodes"], role)
|
||||
return node["meta_data"] if node else {}
|
||||
|
||||
|
||||
def create_serialization_metadata(serialization_scheme, data_source: str,
|
||||
@@ -46,6 +55,10 @@ def create_serialization_metadata(serialization_scheme, data_source: str,
|
||||
"""Creates dict that specifies the serialization metadata."""
|
||||
|
||||
# retrieve parameter from LinkSimulator config
|
||||
# if serialization_scheme empty, return empty dict, It is a meta-data only message
|
||||
if not serialization_scheme:
|
||||
return {}
|
||||
|
||||
if data_source == "5gnr_oai":
|
||||
num_ofdm_symbol = link_sim_parameters["nr_of_symbols"]
|
||||
num_subcarriers = link_sim_parameters["rb_size"] * 12
|
||||
@@ -72,131 +85,73 @@ def create_serialization_metadata(serialization_scheme, data_source: str,
|
||||
|
||||
return serialization_dict
|
||||
|
||||
|
||||
def map_metadata_to_sigmf_format(scope, waveform_generator, parameter_map_file, captured_data):
|
||||
"""
|
||||
Maps metadata from Waveform creator to API and SigMF format.
|
||||
The used parameters and the mapping pairs are specified in a separate YAML file
|
||||
that must be provided as well.
|
||||
"""
|
||||
# read waveform parameter map from yaml file
|
||||
dir_path = os.path.dirname(__file__)
|
||||
src_path = os.path.split(dir_path)[0]
|
||||
|
||||
with open(os.path.join(src_path, parameter_map_file), "r") as file:
|
||||
parameter_map_dic = yaml.load(file, Loader=yaml.Loader)
|
||||
# preallocate target dict
|
||||
sigmf_metadata_dict = {}
|
||||
|
||||
# get standard and name of generator
|
||||
standard_key = parameter_map_dic["waveform_generator"][waveform_generator]
|
||||
generator = standard_key["generator"]
|
||||
|
||||
if scope == "tx":
|
||||
parameter_map_dic = parameter_map_dic["transmitter"][
|
||||
STANDARDS[waveform_generator]
|
||||
]
|
||||
elif scope == "channel":
|
||||
parameter_map_dic = parameter_map_dic["channel"]
|
||||
elif scope == "rx":
|
||||
parameter_map_dic = parameter_map_dic["receiver"]
|
||||
else:
|
||||
raise Exception(
|
||||
f"Invalid mapping scope '{scope}'! Only 'tx', 'channel' and 'rx' are valid!"
|
||||
)
|
||||
|
||||
# check if standard key is given
|
||||
if parameter_map_dic is None:
|
||||
raise Exception(
|
||||
"Invalid standard key: "
|
||||
"Name should be corrected or added to wireless_link_parameter_map.yaml, given: "
|
||||
f"{waveform_generator}"
|
||||
)
|
||||
|
||||
for parameter_pair in parameter_map_dic:
|
||||
# check if key for chosen simulator even exists
|
||||
if waveform_generator + "_parameter" in parameter_pair.keys():
|
||||
# only continue with mapping from file if direct equivalent exists
|
||||
if parameter_pair[waveform_generator + "_parameter"]["name"]:
|
||||
# It is not necessary to get all parameters from wireless_link_parameter_map.yaml
|
||||
# in captured_data since some parameters related to DL or UL only
|
||||
if (parameter_pair[waveform_generator + "_parameter"]["name"] in captured_data.keys()):
|
||||
# extract value from waveform config source
|
||||
value = captured_data[
|
||||
parameter_pair[waveform_generator + "_parameter"]["name"]]
|
||||
# additional mapping if parameter values should come from a discrete set of values
|
||||
if ("value_map" in parameter_pair[waveform_generator + "_parameter"].keys()):
|
||||
value = parameter_pair[waveform_generator + "_parameter"]["value_map"][value]
|
||||
# write to target dictionary for SigMF
|
||||
sigmf_metadata_dict[parameter_pair["sigmf_parameter_name"]] = value
|
||||
else:
|
||||
raise Exception(
|
||||
f"Incomplete specification in field '{waveform_generator}_parameter'!")
|
||||
# else: # fill_non_explicit_fields
|
||||
# waveform_config[parameter_pair["sigmf_parameter_name"]] = "none"
|
||||
if not sigmf_metadata_dict:
|
||||
raise Exception(
|
||||
"""ERROR: Check captured meta-data or provided config and meta data """)
|
||||
|
||||
# check for non-JSON-serializable data types
|
||||
def isfloat(NumberString):
|
||||
try:
|
||||
float(NumberString)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
for key, value in sigmf_metadata_dict.items():
|
||||
if isinstance(value, np.integer):
|
||||
sigmf_metadata_dict[key] = int(value)
|
||||
elif isinstance(value, (np.float16, np.float32, np.float64)):
|
||||
sigmf_metadata_dict[key] = np.format_float_positional(value, trim="-")
|
||||
elif isinstance(value, int):
|
||||
sigmf_metadata_dict[key] = int(value)
|
||||
elif isinstance(value, float):
|
||||
# store value in decimal and not in scientific notation
|
||||
sigmf_metadata_dict[key] = float(value)
|
||||
elif isinstance(value, str) and key != "standard":
|
||||
# convert string to lower case
|
||||
sigmf_metadata_dict[key] = value.lower()
|
||||
if isfloat(value):
|
||||
if value.isdigit():
|
||||
sigmf_metadata_dict[key] = int(float(value))
|
||||
elif value.replace(".", "", 1).isdigit() and value.count(".") < 2:
|
||||
sigmf_metadata_dict[key] = float(value)
|
||||
|
||||
return sigmf_metadata_dict, generator
|
||||
|
||||
|
||||
def create_system_components_metadata(waveform_generator, parameter_map_file, captured_data):
|
||||
# map OAI config data to Wireless Dictionary Parameter Map - SigMF metadata
|
||||
def create_system_components_metadata(captured_data, config_meta_data):
|
||||
"""Creates system components (TX, channel, RX) metadata that will reside in the annotations."""
|
||||
# map metadata of waveform generator to SigMF format
|
||||
signal_info, generator = map_metadata_to_sigmf_format(
|
||||
"tx", waveform_generator, parameter_map_file, captured_data
|
||||
)
|
||||
|
||||
# map OAI config data to Wireless Dictionary Parameter Map - SigMF metadata
|
||||
waveform_generator = config_meta_data["data_recording_config"]["global_info"][
|
||||
"waveform_generator"]
|
||||
parameter_map_file = config_meta_data["data_recording_config"]["parameter_map_file"]
|
||||
|
||||
# map metadata of waveform generator to Wireless Dictionary Parameter Map - SigMF metadata
|
||||
signal_metadata, generator = map_waveform_metadata_to_wireless_dic(
|
||||
"tx", waveform_generator, parameter_map_file, captured_data)
|
||||
|
||||
# derive remaining Signal metadata
|
||||
signal_metadata = derive_remaining_5gnr_metadata(captured_data, signal_metadata)
|
||||
|
||||
tx_metadata = {
|
||||
"signal:detail": {
|
||||
"standard": STANDARDS[waveform_generator],
|
||||
"generator": generator,
|
||||
STANDARDS[waveform_generator]: signal_info,
|
||||
STANDARDS[waveform_generator]: signal_metadata,
|
||||
}
|
||||
}
|
||||
|
||||
channel_metadata = {} # will be filled later
|
||||
rx_metadata = {} # will be filled later
|
||||
|
||||
return tx_metadata, channel_metadata, rx_metadata
|
||||
# get meta data from config file
|
||||
base_station_meta_data = _get_node_meta_data(config_meta_data, "gnb")
|
||||
user_equipment_meta_data = _get_node_meta_data(config_meta_data, "ue")
|
||||
|
||||
# Set number of antennas based on link direction and captured side
|
||||
# Find number of antennas from global info on other side of the link
|
||||
if "UL" in captured_data["message_type"]:
|
||||
# On UE side, It is TX antennas
|
||||
# On gNB side, It is RX antennas
|
||||
if "UE" in captured_data["message_type"]:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"num_tx_antennas"] = captured_data["nb_antennas"]
|
||||
rx_metadata.update({"num_rx_antennas": base_station_meta_data["num_rx_antennas"]})
|
||||
else:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"num_tx_antennas"] = user_equipment_meta_data["num_tx_antennas"]
|
||||
rx_metadata.update({"num_rx_antennas": captured_data["nb_antennas"]})
|
||||
|
||||
if "DL" in captured_data["message_type"]:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"link_direction"] = "downlink"
|
||||
# On UE side, It is RX antennas
|
||||
# On gNB side, It is TX antennasr
|
||||
if "UE" in captured_data["message_type"]:
|
||||
rx_metadata["signal:detail"][STANDARDS[waveform_generator]]["num_rx_antennas"] = \
|
||||
captured_data["nb_antennas"] ## TO DO check key name: no DL message from UE yet
|
||||
tx_metadata.update({"num_tx_antennas": base_station_meta_data["num_tx_antennas"]})
|
||||
else:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"num_tx_antennas"] = captured_data["nrOfLayers"]
|
||||
rx_metadata.update({"num_rx_antennas": user_equipment_meta_data["num_rx_antennas"]})
|
||||
return tx_metadata, channel_metadata, rx_metadata
|
||||
|
||||
def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, idx):
|
||||
"""
|
||||
Compiles and saves provided data and metadata into SigMF file format.
|
||||
"""
|
||||
# get meta data from config file
|
||||
base_station_meta_data = config_meta_data["data_recording_config"]["base_station"][
|
||||
"meta_data"]
|
||||
user_equipment_meta_data = config_meta_data["data_recording_config"][
|
||||
"user_equipment"]["meta_data"]
|
||||
base_station_meta_data = _get_node_meta_data(config_meta_data, "gnb")
|
||||
user_equipment_meta_data = _get_node_meta_data(config_meta_data, "ue")
|
||||
|
||||
# Check the receive target path is valid, else create folder
|
||||
data_storage_path = config_meta_data["data_recording_config"]["data_storage_path"]
|
||||
@@ -206,111 +161,34 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
|
||||
# Write recorded data to file
|
||||
# Get time stamp
|
||||
time_stamp_ms, time_stamp_ms_file_name = time_stamp_formating(
|
||||
captured_data["time_stamp"], global_info["datetime_offset"])
|
||||
time_stamp, time_stamp_file_name = time_stamp_formating(
|
||||
captured_data["unix_capture_ts_sec"], captured_data["unix_capture_ts_nsec"],
|
||||
global_info["datetime_offset"])
|
||||
# Map OAI Message Name to SigMF Message Name
|
||||
file_name_prefix = config_meta_data["data_recording_config"][
|
||||
"supported_oai_tracer_messages"][captured_data["message_type"]]["file_name_prefix"]
|
||||
recorded_data_file_name = (
|
||||
file_name_prefix + "-rec-" + str(idx) + "-" + time_stamp_ms_file_name)
|
||||
file_name_prefix + "-rec-" + str(idx) + "-" + time_stamp_file_name)
|
||||
|
||||
dataset_filename = recorded_data_file_name + ".sigmf-data"
|
||||
dataset_file_path = os.path.join(data_storage_path, dataset_filename)
|
||||
print(dataset_file_path)
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(dataset_file_path)
|
||||
captured_data["recorded_data"].tofile(dataset_file_path)
|
||||
|
||||
# map OAI config data to SigMF metadata
|
||||
# ----------------------------------------------------
|
||||
# map OAI config data to Wireless Dictionary Parameter Map - SigMF metadata
|
||||
waveform_generator = config_meta_data["data_recording_config"]["global_info"][
|
||||
"waveform_generator"]
|
||||
parameter_map_file = config_meta_data["data_recording_config"]["parameter_map_file"]
|
||||
tx_metadata, channel_metadata, rx_metadata = create_system_components_metadata(
|
||||
waveform_generator, parameter_map_file, captured_data)
|
||||
|
||||
# Add other OAI metadata to SigMF metadata
|
||||
# ----------------------------------------------------
|
||||
# Set 1: Parameters needs to be derived from OAI message
|
||||
# ----------------------------------------------------
|
||||
# PUSCH DMRS: OFDM symbol start index within slot
|
||||
get_pusch_dmrs_start_ofdm_symbol = 0
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch_dmrs:ofdm_symbol_idx"] = []
|
||||
|
||||
for symbol in range(captured_data["start_symbol_index"],
|
||||
captured_data["start_symbol_index"] + captured_data["nr_of_symbols"]):
|
||||
|
||||
dmrs_symbol_flag = (captured_data["ul_dmrs_symb_pos"] >> symbol) & 0x01
|
||||
if dmrs_symbol_flag and not get_pusch_dmrs_start_ofdm_symbol:
|
||||
get_pusch_dmrs_start_ofdm_symbol = 1
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch_dmrs:start_ofdm_symbol"] = symbol
|
||||
if dmrs_symbol_flag:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch_dmrs:ofdm_symbol_idx"].append(symbol) # Append symbol to the list
|
||||
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch_dmrs:duration_num_ofdm_symbols"] = 1
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch_dmrs:num_add_positions"] = (captured_data["number_dmrs_symbols"] - 1)
|
||||
|
||||
# If the message is a BIT message, add number of bits to the signal info
|
||||
# "UE_PHY_UL_SCRAMBLED_TX_BITS", "GNB_PHY_UL_PAYLOAD_RX_BITS", "UE_PHY_UL_PAYLOAD_TX_BITS"
|
||||
if "_BITS" in captured_data["message_type"]:
|
||||
# if "BIT" in captured_data["message_type"]:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]]["num_bits"] = (
|
||||
captured_data["number_of_bits"])
|
||||
|
||||
# If the message is captured from UE, it is UPLink message,
|
||||
# so number of antennas is num of Tx antennas
|
||||
# number of antennas on Rx side should be read from global_info given by user
|
||||
if "UE" in captured_data["message_type"]:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"num_tx_antennas"
|
||||
] = captured_data["nb_antennas"]
|
||||
rx_metadata.update(
|
||||
{"num_rx_antennas": base_station_meta_data["num_rx_antennas"]}
|
||||
)
|
||||
# If the message is captured from gNB, it is UPLink message,
|
||||
# so number of antennas is num of Rx antennas
|
||||
# number of antennas on Tx side should be read from global_info given by user
|
||||
else:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"num_tx_antennas"] = user_equipment_meta_data["num_tx_antennas"]
|
||||
rx_metadata.update({"num_rx_antennas": captured_data["nb_antennas"]})
|
||||
|
||||
# Check if the message type contains "UL"
|
||||
if "UL" in captured_data["message_type"]:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"link_direction"] = "uplink"
|
||||
else:
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"link_direction"] = "downlink"
|
||||
# ----------------------------------------------------
|
||||
# Set 2: Parameters that is hardcoded
|
||||
# ----------------------------------------------------
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch:content"] = "compliant"
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch:mapping_type"] = "B"
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]]["num_slots"] = 1
|
||||
if config_meta_data["test_config"]["test_mode"] == "rf_simulation":
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch:payload_bit_pattern"] = "random"
|
||||
elif config_meta_data["test_config"]["test_mode"] == "rf_real_time":
|
||||
tx_metadata["signal:detail"][STANDARDS[waveform_generator]][
|
||||
"pusch:payload_bit_pattern"] = "zeros"
|
||||
else:
|
||||
raise Exception("ERROR: Test mode is not supported in SigMF formatting")
|
||||
tx_metadata, channel_metadata, rx_metadata = create_system_components_metadata(captured_data, config_meta_data)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Read mean parameters
|
||||
freq = config_meta_data["environment_emulation"]["target_link_config"][
|
||||
"wireless_channel"]["carrierFreqValueList_hz"][0]
|
||||
bandwidth = config_meta_data["data_recording_config"]["common_meta_data"][
|
||||
"bandwidth"]
|
||||
# used_signal_bandwidth = (
|
||||
# pow(10, 6) * config_meta_data["environment_emulation"]["target_link_config"]["ran_config"][
|
||||
# "uplink"]["ul_used_signal_BW_MHz"])
|
||||
sample_rate = config_meta_data["data_recording_config"]["common_meta_data"][
|
||||
"sample_rate"]
|
||||
sample_rate = config_meta_data["data_recording_config"]["common_meta_data"]["sample_rate"]
|
||||
|
||||
# get signal emitter info
|
||||
signal_emitter = {
|
||||
@@ -322,23 +200,14 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
"sample_rate": sample_rate,
|
||||
"bandwidth": bandwidth,
|
||||
"gain_tx": user_equipment_meta_data["tx_gain"],
|
||||
"clock_reference": config_meta_data["data_recording_config"][
|
||||
"common_meta_data"
|
||||
]["clock_reference"],
|
||||
"clock_reference": config_meta_data["data_recording_config"]["common_meta_data"]["clock_reference"],
|
||||
}
|
||||
|
||||
# ----------------------------------------------------
|
||||
# get channel metadata
|
||||
# ----------------------------------------------------
|
||||
# get downlink channel info
|
||||
# ch_downlink_config = config_meta_data["environment_emulation"]["target_link_config"]["wireless_channel"]["downlink"]
|
||||
# ch_downlink ={"channel_model": ch_downlink_config["type"]}
|
||||
|
||||
# ch_downlink_model = {"channel_model": "AWGN"}
|
||||
# get uplink channel info
|
||||
ch_uplink_config = config_meta_data["environment_emulation"]["target_link_config"][
|
||||
"wireless_channel"
|
||||
]["uplink"]
|
||||
ch_uplink_config = config_meta_data["environment_emulation"]["target_link_config"]["wireless_channel"]["uplink"]
|
||||
# Real Time RF Emulation
|
||||
if config_meta_data["test_config"]["test_mode"] == "rf_real_time":
|
||||
channel_metadata.update({"emulation_mode": "ni rf real time"})
|
||||
@@ -346,7 +215,7 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
channel_metadata.update(
|
||||
{
|
||||
"channel_model": ch_uplink_config["statistical"]["predef_channel_profile"],
|
||||
"snr_esn0_db": ch_uplink_config["statistical"]["snr_db"],
|
||||
"snr_esn0_db": ch_uplink_config["statistical"]["snr_db"],
|
||||
}
|
||||
)
|
||||
|
||||
@@ -361,6 +230,7 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
)
|
||||
else:
|
||||
raise Exception("ERROR: channel type is not supported in SigMF formatting")
|
||||
|
||||
# RF Simulation
|
||||
elif config_meta_data["test_config"]["test_mode"] == "rf_simulation":
|
||||
channel_metadata.update({"emulation_mode": "oai rf simulation"})
|
||||
@@ -397,16 +267,6 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
],
|
||||
}
|
||||
)
|
||||
# ----------------------------------------------------
|
||||
# get test config
|
||||
# ----------------------------------------------------
|
||||
"""
|
||||
test_config = {
|
||||
"test_name": config_meta_data["test_config"]["test_name"],
|
||||
"test_mode": config_meta_data["test_config"]["test_mode"],
|
||||
"dut_type": config_meta_data["test_config"]["dut_type"],
|
||||
}
|
||||
"""
|
||||
# Create sigmf metadata
|
||||
# ----------------------
|
||||
# Add global parameters to SigMF metadata
|
||||
@@ -439,10 +299,9 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
# Add capture parameters to SigMF metadata
|
||||
# ----------------------
|
||||
serialization_scheme = config_meta_data["data_recording_config"][
|
||||
"supported_oai_tracer_messages"
|
||||
][captured_data["message_type"]]["serialization_scheme"]
|
||||
"supported_oai_tracer_messages"][captured_data["message_type"]]["serialization_scheme"]
|
||||
capture_metadata = {
|
||||
sigmf.SigMFFile.DATETIME_KEY: time_stamp_ms,
|
||||
sigmf.SigMFFile.DATETIME_KEY: time_stamp,
|
||||
SigMFFile.FREQUENCY_KEY: freq,
|
||||
**create_serialization_metadata(
|
||||
serialization_scheme, waveform_generator, captured_data
|
||||
@@ -486,7 +345,8 @@ def write_recorded_data_to_sigmf(captured_data, config_meta_data, global_info, i
|
||||
dataset_meta_file_path = os.path.join(data_storage_path, dataset_meta_filename)
|
||||
meta.tofile(dataset_meta_file_path) # extension is optional
|
||||
|
||||
print(dataset_meta_file_path)
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print(dataset_meta_file_path)
|
||||
|
||||
return dataset_meta_file_path
|
||||
|
||||
@@ -513,8 +373,9 @@ def save_sigmf_collection(streams: list, global_info: dict, description: str, st
|
||||
|
||||
# save collection file
|
||||
collection.tofile(collection_filepath)
|
||||
print("")
|
||||
print(collection_filepath + ".sigmf-collection")
|
||||
if DEBUG_WIRELESS_RECORDED_DATA:
|
||||
print("")
|
||||
print(collection_filepath + ".sigmf-collection")
|
||||
|
||||
|
||||
def load_sigmf(filename: str, storage_path: str, scope: str):
|
||||
@@ -536,7 +397,7 @@ def load_sigmf(filename: str, storage_path: str, scope: str):
|
||||
annotation_length = annotations_metadata[0][sigmf.SigMFFile.LENGTH_INDEX_KEY]
|
||||
|
||||
# get capture metadata
|
||||
capture_metadata = signal.get_capture_info(annotation_start_idx)
|
||||
#capture_metadata = signal.get_capture_info(annotation_start_idx)
|
||||
|
||||
# from source code: sigmffile.py
|
||||
# "autoscale : bool, default True
|
||||
@@ -551,4 +412,4 @@ def load_sigmf(filename: str, storage_path: str, scope: str):
|
||||
data = signal.read_samples(
|
||||
annotation_start_idx, annotation_length, autoscale=False, raw_components=True)
|
||||
|
||||
return data, global_metadata, annotations_metadata
|
||||
return data, global_metadata, annotations_metadata
|
||||
@@ -1,12 +1,10 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief Sync service between captured Data from 5GNR gNB and UE
|
||||
# brief Sync service between captured Data from 5G NR gNB and UE
|
||||
|
||||
import struct
|
||||
from lib import data_recording_messages_def
|
||||
|
||||
DEBUG_POW_MEAS_SYNC = True
|
||||
from lib import shared_memory_interface
|
||||
|
||||
|
||||
# check if first frame ahead:
|
||||
@@ -29,34 +27,55 @@ def is_frame_ahead(frame1, frame2, max_frame=1023):
|
||||
|
||||
|
||||
# find the frame and slot start
|
||||
def find_frame_slot_start(dataset1_start, dataset2_start):
|
||||
def find_frame_slot_start(dataset1_start_info, dataset2_start_info, is_dataset2_pow_monitoring):
|
||||
"""
|
||||
Function to find the frame and slot start for data sync
|
||||
Args:
|
||||
dataset1_start_info: Dictionary containing the start information of dataset1
|
||||
dataset2_start_info: Dictionary containing the start information of dataset2
|
||||
is_dataset2_pow_monitoring: Boolean indicating if dataset2 is pow monitoring
|
||||
Returns:
|
||||
sync_info: Dictionary containing the sync information
|
||||
sync_info["frame"] = frame_start
|
||||
sync_info["slot"] = slot_start
|
||||
Note: if is_dataset2_pow_monitoring is true, then the slot start is taken from dataset1
|
||||
Since the pow monitoring is capturing all slots while the gNB and UE are capturing only data slots
|
||||
|
||||
"""
|
||||
sync_info = {}
|
||||
|
||||
if dataset1_start["frame"] == dataset2_start["frame"]:
|
||||
if dataset1_start_info["frame"] == dataset2_start_info["frame"] :
|
||||
frame_diff = 0
|
||||
frame_start = dataset1_start["frame"]
|
||||
slot_start = max(dataset1_start["slot"], dataset2_start["slot"])
|
||||
frame_start = dataset1_start_info["frame"]
|
||||
if is_dataset2_pow_monitoring:
|
||||
if dataset1_start_info["slot"] >= dataset2_start_info["slot"]:
|
||||
slot_start = dataset1_start_info["slot"]
|
||||
elif dataset1_start_info["slot"] < dataset2_start_info["slot"]:
|
||||
# Go to next frame and use data slot as reference
|
||||
frame_start = (frame_start + 1) % 1024
|
||||
# Use the slot from dataset1 as reference
|
||||
slot_start = dataset1_start_info["slot"]
|
||||
else:
|
||||
slot_start = max(dataset1_start_info["slot"], dataset2_start_info["slot"])
|
||||
|
||||
elif is_frame_ahead(dataset1_start["frame"], dataset2_start["frame"]):
|
||||
frame_start = dataset1_start["frame"]
|
||||
slot_start = dataset1_start["slot"]
|
||||
frame_diff = (dataset1_start["frame"]- dataset2_start["frame"] + 1024) % 1024
|
||||
elif is_frame_ahead(dataset2_start["frame"], dataset1_start["frame"]):
|
||||
frame_start = dataset2_start["frame"]
|
||||
slot_start = dataset2_start["slot"]
|
||||
frame_diff = (dataset2_start["frame"] - dataset1_start["frame"] + 1024) % 1024
|
||||
# check first if the delta time between the two datasets is larger than expected offset time.
|
||||
elif is_frame_ahead(dataset1_start_info["frame"], dataset2_start_info["frame"]):
|
||||
frame_start = dataset1_start_info["frame"]
|
||||
slot_start = dataset1_start_info["slot"]
|
||||
frame_diff = (dataset1_start_info["frame"]- dataset2_start_info["frame"] + 1024) % 1024
|
||||
elif is_frame_ahead(dataset2_start_info["frame"], dataset1_start_info["frame"]):
|
||||
frame_start = dataset2_start_info["frame"]
|
||||
if is_dataset2_pow_monitoring:
|
||||
if dataset1_start_info["slot"] >= dataset2_start_info["slot"]:
|
||||
slot_start = dataset1_start_info["slot"]
|
||||
elif dataset1_start_info["slot"] < dataset2_start_info["slot"]:
|
||||
# Go to next frame and use data slot as reference
|
||||
frame_start = (frame_start + 1) % 1024
|
||||
# Use the slot from dataset1 as reference
|
||||
slot_start = dataset1_start_info["slot"]
|
||||
else:
|
||||
slot_start = dataset2_start_info["slot"]
|
||||
frame_diff = (dataset2_start_info["frame"] - dataset1_start_info["frame"] + 1024) % 1024
|
||||
# check first if the delta time between the two datasets is larger than expected offset time.
|
||||
# So, the sync will not be applied
|
||||
# check during the calculation of the frame the ramp-around from 1023 to 0
|
||||
if abs(frame_diff) > 6:
|
||||
@@ -70,54 +89,34 @@ def find_frame_slot_start(dataset1_start, dataset2_start):
|
||||
sync_info["slot"] = slot_start
|
||||
return sync_info
|
||||
|
||||
# Sync data across N NR tracer nodes
|
||||
def sync_multiple_nr_nodes(node_shm_readings):
|
||||
"""Sync frame/slot across N NR tracer nodes.
|
||||
|
||||
# Read data from gNB T-tracer Application
|
||||
def get_frame_slot_start(shm_reading, bufIdx, general_msg_header_list,
|
||||
general_message_header_length):
|
||||
buf = shm_reading.read(bufIdx + general_message_header_length)
|
||||
msg_id = struct.unpack("<H", buf[bufIdx: bufIdx + general_msg_header_list.get("msg_id")])[0]
|
||||
bufIdx += general_msg_header_list.get("msg_id")
|
||||
frame = struct.unpack("<H", buf[bufIdx: bufIdx + general_msg_header_list.get("frame")])[0]
|
||||
bufIdx += general_msg_header_list.get("frame")
|
||||
slot = struct.unpack("B", buf[bufIdx: bufIdx + general_msg_header_list.get("slot")])[0]
|
||||
return frame, slot
|
||||
|
||||
|
||||
# Sync data between gNB and UE
|
||||
def sync_gnb_ue_captured_data(shm_reading_gnb, shm_reading_ue):
|
||||
"""
|
||||
Function to get the sync (frame, slot) data between gNB and UE
|
||||
Args:
|
||||
shm_reading_gnb: Shared memory for gNB
|
||||
shm_reading_ue: Shared memory for UE
|
||||
Returns:
|
||||
sync_info: Dictionary containing the sync information between gNB and UE
|
||||
sync_info["frame"] = frame_start
|
||||
sync_info["slot"] = slot_start
|
||||
"""
|
||||
# get general message header list
|
||||
general_msg_header_list, general_message_header_length = (
|
||||
data_recording_messages_def.get_general_msg_header_list())
|
||||
node_shm_readings: dict {node_id: shm_reading}
|
||||
|
||||
# Read data from gNB T-tracer Application
|
||||
bufIdx = 0
|
||||
frame_gnb, slot_gnb = get_frame_slot_start(
|
||||
shm_reading_gnb, bufIdx, general_msg_header_list, general_message_header_length)
|
||||
# Read data from UE T-tracer Application
|
||||
bufIdx = 0
|
||||
frame_ue, slot_ue = get_frame_slot_start(
|
||||
shm_reading_ue, bufIdx, general_msg_header_list, general_message_header_length)
|
||||
dataset1_start = {}
|
||||
dataset1_start["frame"] = frame_gnb
|
||||
dataset1_start["slot"] = slot_gnb
|
||||
dataset2_start = {}
|
||||
dataset2_start["frame"] = frame_ue
|
||||
dataset2_start["slot"] = slot_ue
|
||||
# Sync data between gNB and UE
|
||||
# We noticed that the maximum difference between the frame number of gNB and UE is 3 frames
|
||||
# Calculate the frame difference considering the wrap-around from 1023 to 0
|
||||
sync_info = find_frame_slot_start(dataset1_start, dataset2_start)
|
||||
print(" gNB Start info: ", dataset1_start)
|
||||
print(" UE Start info: ", dataset2_start)
|
||||
print(" gNB and UE Sync info: ", sync_info)
|
||||
Returns:
|
||||
sync_info: dict with 'frame' and 'slot' of the common sync point
|
||||
"""
|
||||
sync_header_msg, sync_header_msg_length = (
|
||||
data_recording_messages_def.get_sync_header_msg_list())
|
||||
|
||||
# Get frame/slot from each node
|
||||
node_start_infos = {}
|
||||
for node_id, shm_reading in node_shm_readings.items():
|
||||
frame, slot = shared_memory_interface.get_frame_slot_start(
|
||||
shm_reading, 0, sync_header_msg, sync_header_msg_length)
|
||||
node_start_infos[node_id] = {"frame": frame, "slot": slot}
|
||||
print(f" [NR Node Sync] {node_id} start: frame={frame}, slot={slot}")
|
||||
|
||||
# Find the latest common frame/slot by pairwise comparison
|
||||
node_ids = list(node_start_infos.keys())
|
||||
sync_info = node_start_infos[node_ids[0]]
|
||||
for i in range(1, len(node_ids)):
|
||||
sync_info = find_frame_slot_start(sync_info, node_start_infos[node_ids[i]],
|
||||
is_dataset2_pow_monitoring=False)
|
||||
|
||||
print(f" [NR Node Sync] Multi-node sync point: frame={sync_info['frame']}, slot={sync_info['slot']}")
|
||||
return sync_info
|
||||
|
||||
|
||||
189
common/utils/data_recording/lib/wireless_parameters_mapper.py
Normal file
189
common/utils/data_recording/lib/wireless_parameters_mapper.py
Normal file
@@ -0,0 +1,189 @@
|
||||
# SPDX-License-Identifier: LicenseRef-CSSL-1.0
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# brief Wireless parameters mapper to map captured parameters from waveform generator to API and SigMF format based on a mapping specification in a YAML file.
|
||||
# It also includes functions to derive other parameters that are not directly captured but can be derived from the captured parameters.
|
||||
import os
|
||||
import numpy as np
|
||||
import yaml
|
||||
|
||||
STANDARDS = {"5gnr_oai": "5gnr"}
|
||||
|
||||
|
||||
def map_waveform_metadata_to_wireless_dic(scope, waveform_generator, parameter_map_file, captured_data):
|
||||
"""
|
||||
Maps metadata from Waveform creator to API and SigMF format.
|
||||
The used parameters and the mapping pairs are specified in a separate YAML file
|
||||
that must be provided as well.
|
||||
"""
|
||||
# read waveform parameter map from yaml file
|
||||
dir_path = os.path.dirname(__file__)
|
||||
src_path = os.path.split(dir_path)[0]
|
||||
|
||||
with open(os.path.join(src_path, parameter_map_file), "r") as file:
|
||||
parameter_map_dic = yaml.load(file, Loader=yaml.Loader)
|
||||
# preallocate target dict
|
||||
signal_metadata = {}
|
||||
|
||||
# get standard and name of generator
|
||||
standard_key = parameter_map_dic["waveform_generator"][waveform_generator]
|
||||
generator = standard_key["generator"]
|
||||
|
||||
if scope == "tx":
|
||||
parameter_map_dic = parameter_map_dic["transmitter"][
|
||||
STANDARDS[waveform_generator]
|
||||
]
|
||||
elif scope == "channel":
|
||||
parameter_map_dic = parameter_map_dic["channel"]
|
||||
elif scope == "rx":
|
||||
parameter_map_dic = parameter_map_dic["receiver"]
|
||||
else:
|
||||
raise Exception(
|
||||
f"Invalid mapping scope '{scope}'! Only 'tx', 'channel' and 'rx' are valid!"
|
||||
)
|
||||
|
||||
# check if standard key is given
|
||||
if parameter_map_dic is None:
|
||||
raise Exception(
|
||||
"Invalid standard key: "
|
||||
"Name should be corrected or added to wireless_link_parameter_map.yaml, given: "
|
||||
f"{waveform_generator}"
|
||||
)
|
||||
|
||||
for parameter_pair in parameter_map_dic:
|
||||
# check if key for chosen simulator even exists
|
||||
if waveform_generator + "_parameter" in parameter_pair.keys():
|
||||
# only continue with mapping from file if direct equivalent exists
|
||||
if parameter_pair[waveform_generator + "_parameter"]["name"]:
|
||||
# It is not necessary to get all parameters from wireless_link_parameter_map.yaml
|
||||
# in captured_data since some parameters related to DL or UL only
|
||||
if (parameter_pair[waveform_generator + "_parameter"]["name"] in captured_data.keys()):
|
||||
# extract value from waveform config source
|
||||
value = captured_data[
|
||||
parameter_pair[waveform_generator + "_parameter"]["name"]]
|
||||
# additional mapping if parameter values should come from a discrete set of values
|
||||
if ("value_map" in parameter_pair[waveform_generator + "_parameter"].keys()):
|
||||
value = parameter_pair[waveform_generator + "_parameter"]["value_map"][value]
|
||||
# write to target dictionary for SigMF
|
||||
signal_metadata[parameter_pair["sigmf_parameter_name"]] = value
|
||||
else:
|
||||
raise Exception(
|
||||
f"Incomplete specification in field '{waveform_generator}_parameter'!")
|
||||
if not signal_metadata:
|
||||
raise Exception(
|
||||
"""ERROR: Check captured meta-data or provided config and meta data """)
|
||||
|
||||
# check for non-JSON-serializable data types
|
||||
def isfloat(NumberString):
|
||||
try:
|
||||
float(NumberString)
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
for key, value in signal_metadata.items():
|
||||
if isinstance(value, np.integer):
|
||||
signal_metadata[key] = int(value)
|
||||
elif isinstance(value, (np.float16, np.float32, np.float64)):
|
||||
signal_metadata[key] = np.format_float_positional(value, trim="-")
|
||||
elif isinstance(value, int):
|
||||
signal_metadata[key] = int(value)
|
||||
elif isinstance(value, float):
|
||||
# store value in decimal and not in scientific notation
|
||||
signal_metadata[key] = float(value)
|
||||
elif isinstance(value, str) and key != "standard":
|
||||
# convert string to lower case
|
||||
signal_metadata[key] = value.lower()
|
||||
if isfloat(value):
|
||||
if value.isdigit():
|
||||
signal_metadata[key] = int(float(value))
|
||||
elif value.replace(".", "", 1).isdigit() and value.count(".") < 2:
|
||||
signal_metadata[key] = float(value)
|
||||
|
||||
return signal_metadata, generator
|
||||
|
||||
|
||||
# Derive other Signal metadata from captured metadata
|
||||
def derive_remaining_5gnr_metadata(captured_data, signal_metadata):
|
||||
# Add other OAI metadata to Wireless Dictionary Parameter Map - SigMF metadata
|
||||
# ----------------------------------------------------
|
||||
# Set 1: Parameters needs to be derived from OAI message
|
||||
# ----------------------------------------------------
|
||||
# DMRS Info:
|
||||
# Check if the message type contains "UL"
|
||||
if "UL" in captured_data["message_type"]:
|
||||
# PUSCH DMRS: OFDM symbol start index within slot
|
||||
get_pusch_dmrs_start_ofdm_symbol = 0
|
||||
signal_metadata["pusch_dmrs:ofdm_symbol_idx"] = []
|
||||
|
||||
for symbol in range(captured_data["start_symbol_index"],
|
||||
captured_data["start_symbol_index"] + captured_data["nr_of_symbols"]):
|
||||
|
||||
dmrs_symbol_flag = (captured_data["ul_dmrs_symb_pos"] >> symbol) & 0x01
|
||||
if dmrs_symbol_flag and not get_pusch_dmrs_start_ofdm_symbol:
|
||||
get_pusch_dmrs_start_ofdm_symbol = 1
|
||||
signal_metadata["pusch_dmrs:start_ofdm_symbol"] = symbol
|
||||
if dmrs_symbol_flag:
|
||||
signal_metadata["pusch_dmrs:ofdm_symbol_idx"].append(symbol) # Append symbol to the list
|
||||
|
||||
signal_metadata["pusch_dmrs:duration_num_ofdm_symbols"] = 1
|
||||
signal_metadata["pusch_dmrs:num_add_positions"] = (captured_data["number_dmrs_symbols"] - 1)
|
||||
|
||||
if "DL" in captured_data["message_type"]:
|
||||
# PDSCH DMRS: OFDM symbol start index within slot
|
||||
get_pdsch_dmrs_start_ofdm_symbol = 0
|
||||
signal_metadata["pdsch_dmrs:ofdm_symbol_idx"] = []
|
||||
for symbol in range(captured_data["StartSymbolIndex"],
|
||||
captured_data["StartSymbolIndex"] + captured_data["NrOfSymbols"]):
|
||||
dmrs_symbol_flag = captured_data["dlDmrsSymbPos"] & (1 << symbol)
|
||||
if dmrs_symbol_flag:
|
||||
if dmrs_symbol_flag and not get_pdsch_dmrs_start_ofdm_symbol:
|
||||
get_pdsch_dmrs_start_ofdm_symbol = 1
|
||||
signal_metadata["pdsch_dmrs:start_ofdm_symbol"] = symbol
|
||||
if dmrs_symbol_flag:
|
||||
signal_metadata["pdsch_dmrs:ofdm_symbol_idx"].append(symbol) # Append symbol to the list
|
||||
|
||||
signal_metadata["pdsch_dmrs:duration_num_ofdm_symbols"] = 1
|
||||
signal_metadata["pdsch_dmrs:num_add_positions"] = (captured_data["numDmrsSymbols"] - 1)
|
||||
|
||||
# If the message is a BIT message, add number of bits to the signal info
|
||||
# "UE_PHY_UL_SCRAMBLED_TX_BITS", "GNB_PHY_UL_PAYLOAD_RX_BITS", "UE_PHY_UL_PAYLOAD_TX_BITS"
|
||||
if "_BITS" in captured_data["message_type"]:
|
||||
# if "BIT" in captured_data["message_type"]:
|
||||
signal_metadata["num_bits"] = (captured_data["number_of_bits"])
|
||||
|
||||
# Set Link direction and number of antennas based on link direction and captured side
|
||||
if "UL" in captured_data["message_type"]:
|
||||
signal_metadata["link_direction"] = "uplink"
|
||||
# On UE side, It is TX antennas
|
||||
# On gNB side, It is RX antennas
|
||||
if "UE" in captured_data["message_type"]:
|
||||
signal_metadata["num_tx_antennas"] = captured_data["nb_antennas"]
|
||||
else:
|
||||
signal_metadata["num_rx_antennas"] = captured_data["nb_antennas"]
|
||||
|
||||
if "DL" in captured_data["message_type"]:
|
||||
signal_metadata["link_direction"] = "downlink"
|
||||
# On UE side, It is RX antennas
|
||||
# On gNB side, It is TX antennas
|
||||
if "UE" in captured_data["message_type"]:
|
||||
signal_metadata["num_rx_antennas"] = captured_data["nb_antennas"] ## TO DO check key name: no DL message from UE yet
|
||||
else:
|
||||
signal_metadata["num_tx_antennas"] = captured_data["nrOfLayers"]
|
||||
# ----------------------------------------------------
|
||||
# Set 2: Parameters that is hardcoded
|
||||
# ----------------------------------------------------
|
||||
# Note: The mapping type is hardcoded. It is fixed in OAI to mapping type A for DL and B for UL.
|
||||
# Look to: openair2/RRC/NR/nr_rrc_config.c
|
||||
if "UL" in captured_data["message_type"]:
|
||||
signal_metadata["pusch:mapping_type"] = "B"
|
||||
signal_metadata["pusch:content"] = "compliant"
|
||||
signal_metadata["pusch:payload_bit_pattern"] = "random"
|
||||
if "DL" in captured_data["message_type"]:
|
||||
signal_metadata["pdsch:mapping_type"] = "A"
|
||||
signal_metadata["pdsch:content"] = "compliant"
|
||||
signal_metadata["pdsch:payload_bit_pattern"] = "random"
|
||||
|
||||
signal_metadata["num_slots"] = 1
|
||||
|
||||
return signal_metadata
|
||||
@@ -118,7 +118,7 @@ def calculate_stats(main_config_file: str):
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
main_config = "/home/user/workarea/oai_recorded_data/config_data_recording_app.json"
|
||||
main_config = "/home/user/workarea/oai_recorded_data/config_data_recording_app_extended.json"
|
||||
|
||||
# calculate BER stats
|
||||
calculate_stats(main_config)
|
||||
|
||||
Reference in New Issue
Block a user