Taps client client for vrtsim

Added a libary that can connect to a channel emulator and provide taps
as they are generated.

 - Enable connection from vrtsim to channel emulation server
 - Depends on flatbuffers definition file from raytracing-channel-emulator
 - Depends on libanomsg-dev and flatbuffers-compiler (Ubuntu) or equivalent
 - Added a compile time option ENABLE_TAPS_CLIENT to prevent other users
   from being affected by additional dependencies

openair1/SIMULATION/TOOLS/sim.h

@@ -77,6 +77,8 @@ typedef struct {
   struct complexd **a;
   ///interpolated (sample-spaced) channel impulse response. size(ch) = (n_tx * n_rx) * channel_length. ATTENTION: the dimensions of ch are the transposed ones of a. This is to allow the use of BLAS when applying the correlation matrices to the state.
   struct complexd **ch;
+  ///Same as above but single precision
+  struct complexf **ch_ps;
   ///Sampled frequency response (90 kHz resolution)
   struct complexd **chF;
   ///Maximum path delay in mus.

radio/vrtsim/CMakeLists.txt

@@ -2,3 +2,28 @@ add_library(vrtsim MODULE vrtsim.c noise_device.c)
 set_target_properties(vrtsim PROPERTIES LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
 target_link_libraries(vrtsim PRIVATE SIMU shm_td_iq_channel actor)
 add_dependencies(vrtsim generate_T)
+
+add_boolean_option(OAI_VRTSIM_TAPS_CLIENT OFF "Enable taps client" ON)
+if (OAI_VRTSIM_TAPS_CLIENT)
+  target_link_libraries(vrtsim PRIVATE taps_client)
+  file(DOWNLOAD
+    "https://gitlab.eurecom.fr/oai/raytracing-channel-emulator/-/raw/59589ad2a534881810b2aa7dd81e63066245bb8d/server/api/taps.fbs?inline=false"
+    ${CMAKE_SOURCE_DIR}/taps.fbs
+  )
+
+  set(TAPS_IF_PATH ${CMAKE_SOURCE_DIR}/taps.fbs)
+  set(TAPS_API_HEADER taps_generated.h)
+
+  add_custom_command(OUTPUT ${TAPS_API_HEADER}
+    COMMAND flatc -c ${TAPS_IF_PATH}
+    DEPENDS ${TAPS_IF_PATH}
+    COMMENT "Generating flatbuffers API from ${TAPS_IF_PATH}"
+  )
+  add_library(taps_api INTERFACE)
+  target_include_directories(taps_api INTERFACE ${CMAKE_CURRENT_BINARY_DIR})
+
+  add_library(taps_client taps_client.cpp ${TAPS_API_HEADER})
+  target_include_directories(taps_client PUBLIC .)
+  find_package(Flatbuffers REQUIRED)
+  target_link_libraries(taps_client PUBLIC flatbuffers taps_api SIMU nanomsg)
+endif()

radio/vrtsim/taps_client.cpp

@@ -0,0 +1,198 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+#include <nanomsg/nn.h>
+#include <nanomsg/pubsub.h>
+#include <stdio.h>
+#include <string.h>
+#include <errno.h>
+#include "pthread.h"
+#include "taps_generated.h"
+#include "SIMULATION/TOOLS/sim.h"
+extern "C" {
+#include "assertions.h"
+#include "common/utils/LOG/log.h"
+#include "sim.h"
+#include "utils.h"
+}
+#include <cfloat>
+
+#define NUM_TAPS_BUFFERS 4
+#define MAX_TAPS_LEN 100
+#define MAX_TAPS_MSG_SIZE (sizeof(struct complexf) * MAX_TAPS_LEN * 4 * 4 + 20)
+
+static pthread_t client_thread;
+static bool should_run = true;
+typedef struct {
+  int id;
+  int sock;
+  uint32_t num_tx_antennas;
+  uint32_t num_rx_antennas;
+  channel_desc_t **channel_desc;
+} client_thread_args_t;
+
+typedef struct {
+  void *taps_msg;
+  channel_desc_t *channel_desc;
+} taps_buffer_t;
+
+typedef struct {
+  taps_buffer_t taps_buffers[NUM_TAPS_BUFFERS];
+  int current_buffer;
+} taps_storage_t;
+
+void ascii_line_plot(const float *data, size_t size, char *buffer)
+{
+  const char levels[] = "_.-=#"; // ASCII characters for different levels
+  size_t num_levels = sizeof(levels) - 1; // Number of levels (excluding the null terminator)
+
+  // Calculate min and max values from the data
+  float min_val = FLT_MAX;
+  float max_val = FLT_MIN;
+  for (size_t i = 0; i < size; i++) {
+    if (data[i] < min_val)
+      min_val = data[i];
+    if (data[i] > max_val)
+      max_val = data[i];
+  }
+
+  // Handle edge case where all values are the same
+  if (min_val == max_val) {
+    min_val -= 1.0f;
+    max_val += 1.0f;
+  }
+
+  // Normalize and map data to levels
+  for (size_t i = 0; i < size; i++) {
+    float normalized = (data[i] - min_val) / (max_val - min_val); // Normalize to [0, 1]
+    size_t level_index = (size_t)(normalized * num_levels); // Map to level index
+    if (level_index > num_levels)
+      level_index = num_levels; // Clamp to valid range
+    snprintf(buffer + i, 2, "%c", levels[level_index]);
+  }
+}
+
+static void init_taps_storage(taps_storage_t *storage, int num_tx_antennas, int num_rx_antennas)
+{
+  for (int i = 0; i < NUM_TAPS_BUFFERS; i++) {
+    storage->taps_buffers[i].taps_msg = calloc_or_fail(1, MAX_TAPS_MSG_SIZE);
+    storage->taps_buffers[i].channel_desc = (channel_desc_t *)calloc_or_fail(1, sizeof(channel_desc_t));
+    storage->taps_buffers[i].channel_desc->ch_ps =
+        (struct complexf **)calloc_or_fail(num_rx_antennas * num_tx_antennas, sizeof(struct complexf *));
+  }
+  storage->current_buffer = 0;
+}
+
+static taps_storage_t taps_storage;
+
+void *client_thread_func(void *args)
+{
+  client_thread_args_t *client_thread_args = (client_thread_args_t *)args;
+  while (should_run) {
+    int next_buffer = (taps_storage.current_buffer + 1) % NUM_TAPS_BUFFERS;
+    taps_buffer_t *taps_buffer = &taps_storage.taps_buffers[next_buffer];
+    int ret = nn_recv(client_thread_args->sock, taps_buffer->taps_msg, MAX_TAPS_MSG_SIZE, 0);
+    if (ret < 0) {
+      LOG_E(HW, "nn_recv() failed: errno: %d, %s\n", errno, strerror(errno));
+      continue;
+    }
+    auto taps_message = Phy::GetTaps(taps_buffer->taps_msg);
+    if (taps_message->num_rx_antennas() != client_thread_args->num_rx_antennas
+        || taps_message->num_tx_antennas() != client_thread_args->num_tx_antennas) {
+      LOG_E(HW,
+            "Number of antennas mismatch: expected %d x %d, got %d x %d\n",
+            client_thread_args->num_rx_antennas,
+            client_thread_args->num_tx_antennas,
+            taps_message->num_rx_antennas(),
+            taps_message->num_tx_antennas());
+      continue;
+    }
+    channel_desc_t *channel_desc = taps_buffer->channel_desc;
+    const flatbuffers::Vector<float> *taps = taps_message->taps();
+    struct complexf *base_pointer = (struct complexf *)taps->data();
+    int taps_len = taps_message->taps_len();
+    AssertFatal(taps_len < MAX_TAPS_LEN, "Too many samples in the taps array, taps_len = %d\n", taps_len);
+    for (unsigned int aarx = 0U; aarx < client_thread_args->num_rx_antennas; aarx++) {
+      for (unsigned int aatx = 0U; aatx < client_thread_args->num_tx_antennas; aatx++) {
+        channel_desc->ch_ps[aarx + (client_thread_args->num_rx_antennas * aatx)] =
+            &base_pointer[(aarx + (client_thread_args->num_rx_antennas * aatx)) * taps_len];
+      }
+    }
+    channel_desc->path_loss_dB = 0;
+    channel_desc->channel_length = taps_len;
+    *client_thread_args->channel_desc = channel_desc;
+    taps_storage.current_buffer = next_buffer;
+    LOG_A(HW, "Receved new taps message, channel_length %d, buffer %d\n", channel_desc->channel_length, next_buffer);
+    for (unsigned int aarx = 0; aarx < client_thread_args->num_rx_antennas; aarx++) {
+      for (unsigned int aatx = 0; aatx < client_thread_args->num_tx_antennas; aatx++) {
+        char buffer[MAX_TAPS_LEN + 1];
+        memset(buffer, 0, sizeof(buffer));
+        float magnitudes[MAX_TAPS_LEN];
+        cf_t *channel = channel_desc->ch_ps[aarx + (client_thread_args->num_rx_antennas * aatx)];
+        for (int i = 0; i < channel_desc->channel_length; i++) {
+          magnitudes[i] = sqrtf(powf(channel[i].r, 2) + powf(channel[i].i, 2));
+        }
+        ascii_line_plot(magnitudes, channel_desc->channel_length, buffer);
+        LOG_A(HW, "Taps message %d, channel %d x %d: %s\n", client_thread_args->id, aarx, aatx, buffer);
+      }
+    }
+  }
+  return NULL;
+}
+
+extern "C" void taps_client_connect(int id,
+                                    const char *socket_path,
+                                    int num_tx_antennas,
+                                    int num_rx_antennas,
+                                    channel_desc_t **channel_desc)
+{
+  // Create a socket
+  int sock = nn_socket(AF_SP, NN_SUB);
+  AssertFatal(sock >= 0, "nn_socket() failed: errno: %d, %s\n", errno, strerror(errno));
+
+  int ret = nn_connect(sock, socket_path);
+  AssertFatal(ret >= 0, "nn_connect() failed: errno: %d, %s\n", errno, strerror(errno));
+
+  // Subscribe to all messages
+  ret = nn_setsockopt(sock, NN_SUB, NN_SUB_SUBSCRIBE, "", 0);
+  AssertFatal(ret == 0, "nn_setsockopt() failed, errno %d, %s\n", errno, strerror(errno));
+
+  init_taps_storage(&taps_storage, num_tx_antennas, num_rx_antennas);
+
+  client_thread_args_t *client_thread_args = static_cast<client_thread_args_t *>(malloc(sizeof(client_thread_args_t)));
+  client_thread_args->id = id;
+  client_thread_args->sock = sock;
+  client_thread_args->num_rx_antennas = num_rx_antennas;
+  client_thread_args->num_tx_antennas = num_tx_antennas;
+  client_thread_args->channel_desc = channel_desc;
+  ret = pthread_create(&client_thread, NULL, client_thread_func, client_thread_args);
+  AssertFatal(ret == 0, "pthread_create() failed: errno: %d, %s\n", errno, strerror(errno));
+}
+
+extern "C" void taps_client_stop()
+{
+  should_run = false;
+  pthread_join(client_thread, NULL);
+  for (int i = 0; i < NUM_TAPS_BUFFERS; i++) {
+    free(taps_storage.taps_buffers[i].taps_msg);
+    free(taps_storage.taps_buffers[i].channel_desc->ch_ps);
+    free(taps_storage.taps_buffers[i].channel_desc);
+  }
+}

radio/vrtsim/taps_client.h

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#ifndef TAPS_CLIENT_H
+#define TAPS_CLIENT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "sim.h"
+
+void taps_client_connect(int id, const char *socket_path, int num_tx_antennas, int num_rx_antennas, channel_desc_t **channel_desc);
+void taps_client_stop();
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

radio/vrtsim/vrtsim.c

@@ -47,6 +47,7 @@
 #include "actor.h"
 #include "noise_device.h"
 #include "simde/x86/avx512.h"
+#include "taps_client.h"
 
 // Simulator role
 typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
@@ -60,6 +61,7 @@ typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
 #define VRTSIM_SECTION "vrtsim"
 #define TIME_SCALE_HLP \
   "sample time scale. 1.0 means realtime. Values > 1 mean faster than realtime. Values < 1 mean slower than realtime\n"
+#define TAPS_SOCKET_HLP "Socket to connect to the channel emulation server\n"
 
 // clang-format off
 #define VRTSIM_PARAMS_DESC \
@@ -68,6 +70,7 @@ typedef enum { ROLE_SERVER = 1, ROLE_CLIENT } role;
      {"role",         "either client or server\n",     0, .strptr = &role,                       .defstrval = ROLE_CLIENT_STRING,  TYPE_STRING, 0}, \
      {"timescale",    TIME_SCALE_HLP,                  0, .dblptr = &vrtsim_state->timescale, .defdblval = 1.0,                 TYPE_DOUBLE, 0}, \
      {"chanmod",      "Enable channel modelling",      0, .iptr = &vrtsim_state->chanmod,     .defintval = 0,                   TYPE_INT,    0}, \
+     {"taps-socket",  TAPS_SOCKET_HLP,                 0, .strptr = &vrtsim_state->taps_socket, .defstrval = NULL, TYPE_STRING, 0}, \
   };
 // clang-format on
 
@@ -112,6 +115,7 @@ typedef struct {
   int rx_num_channels;
   channel_desc_t *channel_desc;
   Actor_t *channel_modelling_actors;
+  char *taps_socket;
 } vrtsim_state_t;
 
 // Sample history for channel impulse response
@@ -183,6 +187,15 @@ static void vrtsim_readconfig(vrtsim_state_t *vrtsim_state)
   } else {
     AssertFatal(false, "Invalid role configuration\n");
   }
+#ifdef ENABLE_TAPS_CLIENT
+  if (vrtsim_state->taps_socket) {
+    LOG_A(HW, "VRTSIM: will use taps socket %s\n", vrtsim_state->taps_socket);
+  }
+#else
+  if (vrtsim_state->taps_socket) {
+    AssertFatal(false, "Invalid configuration: Build with ENABLE_TAPS_CLIENT to use taps socket\n");
+  }
+#endif
 }
 
 static void *vrtsim_timing_job(void *arg)
@@ -326,12 +339,20 @@ static int vrtsim_connect(openair0_device *device)
 
   // Handle channel modelling after number of RX antennas are known
   int num_tx_stats = 1;
-  if (vrtsim_state->chanmod) {
+  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
     vrtsim_state->channel_modelling_actors = calloc_or_fail(vrtsim_state->peer_info.num_rx_antennas, sizeof(Actor_t));
     for (int i = 0; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
       init_actor(&vrtsim_state->channel_modelling_actors[i], "chanmod", -1);
     }
-    load_channel_model(vrtsim_state);
+    if (vrtsim_state->taps_socket) {
+      taps_client_connect(0,
+                          vrtsim_state->taps_socket,
+                          device->openair0_cfg[0].tx_num_channels,
+                          vrtsim_state->peer_info.num_rx_antennas,
+                          &vrtsim_state->channel_desc);
+    } else {
+      load_channel_model(vrtsim_state);
+    }
     num_tx_stats = vrtsim_state->peer_info.num_rx_antennas;
   }
   vrtsim_state->tx_timing = calloc_or_fail(num_tx_stats, sizeof(tx_timing_t));
@@ -401,6 +422,7 @@ typedef struct {
 static void perform_channel_modelling(void *arg)
 {
   channel_modelling_args_t *channel_modelling_args = arg;
+  vrtsim_state_t *vrtsim_state = channel_modelling_args->vrtsim_state;
   int nsamps = channel_modelling_args->nsamps;
   int aarx = channel_modelling_args->aarx;
   int nb_tx_ant = channel_modelling_args->nbAnt;
@@ -411,23 +433,36 @@ static void perform_channel_modelling(void *arg)
   // Apply noise from global settings
   get_noise_vector((float *)samples, nsamps * 2);
 
-  channel_desc_t *channel_desc = channel_modelling_args->vrtsim_state->channel_desc;
-  const float pathloss_linear = powf(10, channel_desc->path_loss_dB / 20.0);
+  channel_desc_t *channel_desc = vrtsim_state->channel_desc;
+
+  if (channel_desc == NULL) {
+    return;
+  }
 
-  // Convert channel impulse response to float + apply pathloss
   cf_t channel_impulse_response[nb_tx_ant][channel_desc->channel_length];
-  for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
-    const struct complexd *channelModel = channel_desc->ch[aarx + (aatx * channel_desc->nb_rx)];
-    for (int i = 0; i < channel_desc->channel_length; i++) {
-      channel_impulse_response[aatx][i].r = channelModel[i].r * pathloss_linear;
-      channel_impulse_response[aatx][i].i = channelModel[i].i * pathloss_linear;
+  cf_t *channel_impulse_response_p[nb_tx_ant];
+  if (!vrtsim_state->taps_socket) {
+    const float pathloss_linear = powf(10, channel_desc->path_loss_dB / 20.0);
+    // Convert channel impulse response to float + apply pathloss
+    for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+      const struct complexd *channelModel = channel_desc->ch[aarx + (aatx * channel_desc->nb_rx)];
+      for (int i = 0; i < channel_desc->channel_length; i++) {
+        channel_impulse_response[aatx][i].r = channelModel[i].r * pathloss_linear;
+        channel_impulse_response[aatx][i].i = channelModel[i].i * pathloss_linear;
+      }
+      channel_impulse_response_p[aatx] = channel_impulse_response[aatx];
+    }
+  } else {
+    for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
+      struct complexf *channelModel = channel_desc->ch_ps[aarx + (aatx * channel_desc->nb_rx)];
+      channel_impulse_response_p[aatx] = channelModel;
     }
   }
 
   for (int aatx = 0; aatx < nb_tx_ant; aatx++) {
     c16_t *previous_samples = saved_samples[aatx];
     for (int i = 0; i < nsamps; i++) {
-      cf_t *impulse_response = channel_impulse_response[aatx];
+      cf_t *impulse_response = channel_impulse_response_p[aatx];
       for (int l = 0; l < channel_desc->channel_length; l++) {
         int idx = i - l;
         // TODO: Use AVX2 for this
@@ -515,8 +550,9 @@ static int vrtsim_write(openair0_device *device, openair0_timestamp timestamp, v
 {
   timestamp -= device->openair0_cfg->command_line_sample_advance;
   vrtsim_state_t *vrtsim_state = (vrtsim_state_t *)device->priv;
-  return vrtsim_state->chanmod ? vrtsim_write_with_chanmod(vrtsim_state, timestamp, samplesVoid, nsamps, nbAnt, flags)
-                               : vrtsim_write_internal(vrtsim_state, timestamp, (c16_t *)samplesVoid[0], nsamps, 0, flags, 0);
+  bool channel_modelling = vrtsim_state->chanmod || vrtsim_state->taps_socket;
+  return channel_modelling ? vrtsim_write_with_chanmod(vrtsim_state, timestamp, samplesVoid, nsamps, nbAnt, flags)
+                           : vrtsim_write_internal(vrtsim_state, timestamp, (c16_t *)samplesVoid[0], nsamps, 0, flags, 0);
 }
 
 static int vrtsim_read(openair0_device *device, openair0_timestamp *ptimestamp, void **samplesVoid, int nsamps, int nbAnt)
@@ -545,7 +581,7 @@ static void vrtsim_end(openair0_device *device)
   }
 
   tx_timing_t *tx_timing = vrtsim_state->tx_timing;
-  if (vrtsim_state->chanmod) {
+  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
     for (int i = 0; i < vrtsim_state->peer_info.num_rx_antennas; i++) {
       shutdown_actor(&vrtsim_state->channel_modelling_actors[i]);
     }
@@ -559,6 +595,9 @@ static void vrtsim_end(openair0_device *device)
     }
     tx_timing->average_tx_budget /= vrtsim_state->peer_info.num_rx_antennas;
     free_noise_device();
+    if (vrtsim_state->taps_socket) {
+      taps_client_stop();
+    }
   }
   // produce 1 second of extra samples so threads can finish
   shm_td_iq_channel_produce_samples(vrtsim_state->channel, vrtsim_state->sample_rate);
@@ -621,10 +660,11 @@ __attribute__((__visibility__("default"))) int device_init(openair0_device *devi
   vrtsim_state->tx_num_channels = openair0_cfg->tx_num_channels;
   vrtsim_state->rx_num_channels = openair0_cfg->rx_num_channels;
 
-  if (vrtsim_state->chanmod) {
+  if (vrtsim_state->chanmod || vrtsim_state->taps_socket) {
     init_channelmod();
     int noise_power_dBFS = get_noise_power_dBFS();
     int16_t noise_power = noise_power_dBFS == INVALID_DBFS_VALUE ? 0 : (int16_t)(32767.0 / powf(10.0, .05 * -noise_power_dBFS));
+    LOG_A(HW, "VRTSIM: Noise power %d sample value\n", noise_power);
     init_noise_device(noise_power);
   }
   return 0;