diff --git a/common/utils/data_recording/config/config_data_recording.json b/common/utils/data_recording/config/config_data_recording.json index 2d4387d664..29ebb6bd4c 100644 --- a/common/utils/data_recording/config/config_data_recording.json +++ b/common/utils/data_recording/config/config_data_recording.json @@ -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 + } + } + ] } -} +} \ No newline at end of file diff --git a/common/utils/data_recording/data_recording_app_v1.0.py b/common/utils/data_recording/data_recording_app_v1.0.py deleted file mode 100644 index 978ce13ceb..0000000000 --- a/common/utils/data_recording/data_recording_app_v1.0.py +++ /dev/null @@ -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(' 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(" ({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") diff --git a/common/utils/data_recording/lib/common_utils.py b/common/utils/data_recording/lib/common_utils.py index bbc29a73d2..3d6895245a 100644 --- a/common/utils/data_recording/lib/common_utils.py +++ b/common/utils/data_recording/lib/common_utils.py @@ -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}") diff --git a/common/utils/data_recording/lib/config_interface.py b/common/utils/data_recording/lib/config_interface.py index 29df0b5c42..5a95ffee83 100644 --- a/common/utils/data_recording/lib/config_interface.py +++ b/common/utils/data_recording/lib/config_interface.py @@ -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 diff --git a/common/utils/data_recording/lib/data_recording_messages_def.py b/common/utils/data_recording/lib/data_recording_messages_def.py index 195f58f788..646bdd275d 100644 --- a/common/utils/data_recording/lib/data_recording_messages_def.py +++ b/common/utils/data_recording/lib/data_recording_messages_def.py @@ -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, diff --git a/common/utils/data_recording/lib/shared_memory_interface.py b/common/utils/data_recording/lib/shared_memory_interface.py new file mode 100644 index 0000000000..b4aabdc924 --- /dev/null +++ b/common/utils/data_recording/lib/shared_memory_interface.py @@ -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(" 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("> 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 \ No newline at end of file diff --git a/common/utils/data_recording/lib/sync_service.py b/common/utils/data_recording/lib/sync_service.py index e23aa1a773..ebd2ca7431 100644 --- a/common/utils/data_recording/lib/sync_service.py +++ b/common/utils/data_recording/lib/sync_service.py @@ -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("> 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 diff --git a/common/utils/data_recording/sync_validation_demo.py b/common/utils/data_recording/sync_validation_demo.py index 1097d49498..b3dd43a002 100644 --- a/common/utils/data_recording/sync_validation_demo.py +++ b/common/utils/data_recording/sync_validation_demo.py @@ -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)