Code for wireless communication work

I need some pointers on how I can speed up my code, as of now it is incredible slow for larger inputs. The way the code works is that the file Loc_Circle_50U.txt contains the true locations of 50 vehicles while the files in ns3_files contain some erroneous locations. I analyze these differences which are stored as error, and based on the error and the velocities of the vehicles, I calculate if they are likely to collide. Time is divided into slots of 1 milliseconds.

Testing files are shared here. The sumo_file = ['Loc_Circle_50U.txt'] is a global file which is 25MB when extracted, and the files listed in ns3_files that run one by one. Currently the one I have attached for ns3_files is a smaller one, and the bigger ones are around 30MB.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import math
from joblib import Parallel, delayed

ns3_files = ['EQ_100p_1Hz_50U.txt','EQ_100p_5Hz_50U.txt','EQ_100p_10Hz_50U.txt',
'EQ_100p_20Hz_50U.txt','EQ_100p_30Hz_50U.txt','EQ_100p_40Hz_50U.txt','EQ_100p_50Hz_50U.txt','EQ_100p_100Hz_50U.txt']

sumo_file = ['Loc_Circle_50U.txt']
sumo_df = pd.read_csv(sumo_file[0], delim_whitespace = True)

for qqq in sumo_file:
rr = pd.read_csv(qqq, delim_whitespace=True)
print("analyzing file ", qqq)

if 'Time' in rr.columns:
if 'VID' in rr.columns:
if 'Pos(X)' in rr.columns:
if 'Pos(Y)' in rr.columns:
if 'Vel(X)' in rr.columns:
if 'Vel(Y)' in rr.columns:
print("sumo file ", qqq, " is OK")

for qqq in ns3_files:
rr = pd.read_csv(qqq, delim_whitespace=True)
print("analyzing file ", qqq)

if 'PId' in rr.columns:
if 'TxId' in rr.columns:
if 'RxId' in rr.columns:
if 'GenTime' in rr.columns:
if 'RecTime' in rr.columns:
if 'Delay' in rr.columns:
if 'TxTruePos(X)' in rr.columns:
if 'TxHeadPos(X)' in rr.columns:
if 'TxTruePos(Y)' in rr.columns:
if 'TxHeadPos(Y)' in rr.columns:
if 'Error(m)' in rr.columns:
if 'Tx_HeadVelX' in rr.columns:
if 'Tx_HeadVelY' in rr.columns:
if 'RecvPos(X)' in rr.columns:
if 'RecvPos(Y)' in rr.columns:
print("ns3 file ", qqq, " is OK")

prediction = 0 # 0 means no prediction and 1 is prediction enabled

def calc_error(c): # pass the ns3 dataframe cleaned of all nan values.
c.sort_values("RecTime")
#     print("c = ", c, " to be processed")
nrows = c.shape[0]
error = [] # will store slot wise error
collision = 0 # will be 0 as long as ttc error < 6.43. Becomes 1 if ttc_error exceeds 6.43 even for 1 slot
ttc_error_x = 0 # calculates the ttc error in x for every slot
ttc_error_y = 0 # calculates the ttc error in y for every slot
ttc_error = [] # will store slot wise ttc error
sender = c.loc[0, "TxId"] # sender will be the same throughout, so take the sender value from any row
receiver = c.loc[0, "RxId"] # same as above
#     print("sender = ", sender, "receiver = ", receiver)
if nrows==1: # only 1 message exchanged
error_x_val = abs( (c["TxTruePos(X)"]) - c["TxHeadPos(X)"]).values[0]
error_y_val = abs( (c["TxTruePos(Y)"]) - c["TxHeadPos(Y)"]).values[0]
rel_vel_x = abs (c["Tx_HeadVelX"] - c["RecvVel(X)"]).values[0]
rel_vel_y = abs (c["Tx_HeadVelY"] - c["RecvVel(Y)"]).values[0]
# now for the relative velocity, which of the sender's velocity to take ? the sending instant or receiving instant ?
#         print("error_x = ", error_x, "rel_vel_x = ", rel_vel_x)
if (rel_vel_x!=0):
ttc_error_x = error_x_val/rel_vel_x
else:
ttc_error_x = 0 # rel vel same means no error
if (rel_vel_y!=0):
ttc_error_y = error_y_val/rel_vel_y
else:
ttc_error_y = 0
#         print("1 packet scenario ", ttc_error_x, ttc_error_y, error_x_val, error_y_val, rel_vel_x, rel_vel_y)
ttc_error.append(max(ttc_error_x, ttc_error_y))
err = c["Error(m)"].values[0]
error.append(np.mean(err))

else: # more than 1 packet exchanged

for k in range(nrows-1): # one k for each BSM. here BSMs are analyzed at reception instants
current_bsm = c.loc[k]
next_bsm =  c.loc[k+1]
slots = int(next_bsm['RecTime'] - current_bsm['RecTime'] - 1)
current_time = current_bsm["RecTime"]
mask_1 = (sumo_df['VID'] == sender)
df_actual = sumo_df[mask_1] # df_actual is senders sumo information
# print("current_bsm" , current_bsm) #["RecTime"]) #["RecvVel(X)"])
x_actual=(current_bsm["TxTruePos(X)"])
y_actual=(current_bsm["TxTruePos(Y)"])
error_x_val = abs(x_actual - x_header)
error_y_val = abs(y_actual - y_header)
error.append(math.sqrt(error_x_val**2 + error_y_val**2))
rel_vel_x = abs(sender_velocity_x - current_bsm["RecvVel(X)"])
rel_vel_y = abs(sender_velocity_y - current_bsm["RecvVel(Y)"])
#             print("sender_velocity_x=",sender_velocity_x,"sender_velocity_y=",sender_velocity_y,"rec_vel_x=",current_bsm["RecvVel(X)"],"rec_vel_y",current_bsm["RecvVel(Y)"], \
#                  "rel_vel_x",rel_vel_x,"rel_vel_y",rel_vel_y)
# the next are header info as error is from rx perspective and rx has only header info
x_pos_BSM = c.loc[k, "TxHeadPos(X)"]
y_pos_BSM = c.loc[k, "TxHeadPos(Y)"]
x_speed_BSM = c.loc[k, "Tx_HeadVelX"]
y_speed_BSM = c.loc[k, "Tx_HeadVelY"]

if (rel_vel_x!=0):
ttc_error_x = error_x_val/rel_vel_x
else:
ttc_error_x = 0
if (rel_vel_y!=0):
ttc_error_y = error_y_val/rel_vel_y
else:
ttc_error_y = 0
ttc_error.append(max(ttc_error_x, ttc_error_y))
#             print(" BSM at", time," rel_vel_x=",rel_vel_x,"rel_vel_y=",rel_vel_y,"error_x=",error_x,"error_y=",error_y  )
#             print("ttc_error_x=", ttc_error_x, "ttc_error_y=", ttc_error_y)
for j in range(slots): # this for loop will run fir every slot in between 2 receptions
#                 print("prediction slot = ", current_slot+j+1)
x_pos_predicted = x_pos_BSM + prediction*(x_speed_BSM * (j+1))*(0.001) # as slots are in msec and speed in m/s
y_pos_predicted = y_pos_BSM + prediction*(y_speed_BSM * (j+1))*(0.001)
mask_3 = (df_actual["Time"] == (current_time + (j+1))) # df_actual has the senders info
#                 print(df_row)
# row of sumo file at the ongoing slot for receiver
x_pos_actual = df_row["Pos(X)"].values[0]
y_pos_actual = df_row["Pos(Y)"].values[0]
#                 print("x actual=", x_pos_actual," y actual=",y_pos_actual," x pred=",x_pos_predicted, " y pred =", y_pos_predicted)
error_x_val = abs((x_pos_predicted) - (x_pos_actual))
error_y_val = abs((y_pos_predicted) - (y_pos_actual))

error.append(error_x_val**2 + error_y_val**2)
#                 print("x error = ", error_x, ", y error = ", error_y)
df_receiver = sumo_df[receiver_mask] # the row for sender at that instant
rel_vel_x = abs(df_row["Vel(X)"].values[0] - df_receiver["Vel(X)"].values[0])
rel_vel_y = abs(df_row["Vel(Y)"].values[0] - df_receiver["Vel(Y)"].values[0])
if (rel_vel_x!=0):
ttc_error_x = error_x_val/rel_vel_x
else:
ttc_error_x = 0
if (rel_vel_y!=0):
ttc_error_y = error_y_val/rel_vel_y
else:
ttc_error_y = 0
ttc_error.append(max(ttc_error_x, ttc_error_y))
#                 print("ttc_error_x=", ttc_error_x, "ttc_error_y=", ttc_error_y)
#                 print("predslot",current_time+j+1,"x_actual",x_pos_actual,"y_actual",y_pos_actual,"x_predicted",x_pos_predicted,"y_predicted",y_pos_predicted,"error_x_val",error_x_val,"error_y_val",error_y_val) #, " is ", slot_error)

# add the last packet details
err_lastpacket = c.loc[nrows-1, "Error(m)"]
error.append(err_lastpacket)
current_time = c.loc[nrows-1, "RecTime"]
error_x_val = abs( (c.loc[nrows-1,"TxTruePos(X)"]) - c.loc[nrows-1,"TxHeadPos(X)"])
error_y_val = abs( (c.loc[nrows-1,"TxTruePos(Y)"]) - c.loc[nrows-1,"TxHeadPos(Y)"])
sender_mask = ((sumo_df["VID"]==sender) & (sumo_df["Time"]==(current_time)))
rel_vel_x = abs (sender_x_vel - c.loc[nrows-1,"RecvVel(X)"])
rel_vel_y = abs (sender_y_vel - c.loc[nrows-1,"RecvVel(Y)"])
if (rel_vel_x!=0):
ttc_error_x = error_x_val/rel_vel_x
#         print("error_x_val",error_x_val,"rel_vel_x",rel_vel_x)
else:
ttc_error_x = 0
if (rel_vel_y!=0):
ttc_error_y = error_y_val/rel_vel_y
else:
ttc_error_y = 0
#     print("ttc_error_x",ttc_error_x, "ttc_error_y",ttc_error_y)
ttc_error.append(max(ttc_error_x, ttc_error_y))

#             print("current_time",current_time,"sender ",sender)
#             print("overall error", error)
#     print("overall ttc_error", ttc_error)
#     print("\n")
#         print("error for sender", sender, "and receiver", receiver, "is", error)
avg_error = np.mean(error)
if np.mean(ttc_error)>6.43:
collision = 1
else:
collision = 0
return (avg_error, collision)

overall_errors = [] # to store error per file
overall_collisions = [] # to store collision per file

def start_process(fil):
print("File ", fil, " started")
b = pd.read_csv(fil, delim_whitespace = True)
b = b.sort_values(['RecTime'])
b = b.reset_index(drop=True)
m = b['RxId'].nunique() # m is number of receivers
# throughput block starts
#     overall_duration = (b['RecTime'].max() - b['RecTime'].min())/1000 # milliseconds to seconds

## in throughput case, we work on the whole file so 1 pair or 1 packet exchanged cases do not apply.

#     packets = b.shape[0] # no of rows = no of packets received

average_errors = [] # hold error for every pair in a file
average_collisions = [] # hold collision (0 or 1) for every pair in a file
#     collisions = 0 # will have the number of collision risky pairs in every file
for i in range(len(receivers)):
senders = b[b['RxId'] == receiver].TxId.unique()
for j in range(len(senders)):
sender = senders[j]
mask = (b['RxId'] == receiver) & (b['TxId'] == sender) # extract out the rx-tx pair
#             print("cc=",cc)
c = c.reset_index(drop=True)
#             print("cc=",cc)
#             print("error calculation for sender ",sender, " and receiver ", receiver, "\n")
#             print("c = ", c , "before being sent")
avg_error, collision = calc_error(c) # calc_error is the function
# this will give the whole error for that pair
# pos_error should return a value of error
#             avg_error = np.sum(pos_error)/overall_duration # errors for single pair
average_errors.append(avg_error) # average_errors will hold the error for every pair in a file
average_collisions.append(collision)
#                 print("average errors for Tx ",sender, " and receiver ", receiver, " is ", avg_error, "\n")
#             print("collision status is ", collision)
#         print("average_collisions", average_collisions,"average_errors",average_errors)
average_error = np.average(average_errors)
average_collision = np.average(average_collisions)

print("File ", fil, " completed")

#     print("\n")
#     print("average_error = ", average_error)
overall_collisions.append(average_collision)
overall_errors.append(average_error)
# print(average_errors)

if prediction==0:
print("for file ", fil, file = open("parallel_error_collision_P.txt", "a"))
print("no prediction result follows with prediction flag =", prediction, file = open("parallel_error_collision_P.txt", "a"))
print("overall_collisions = ", overall_collisions, file = open("parallel_error_collision_P.txt", "a"))
print("overall_errors = ", overall_errors,"\n", file = open("parallel_error_collision_P.txt", "a"))
else:
print("for file ", fil, file = open("parallel_error_collision_P.txt", "a"))
print("prediction assisted result follows with prediction flag =", prediction, file = open("parallel_error_collision_P.txt", "a"))
print("overall_collisions = ", overall_collisions, file = open("parallel_error_collision_P.txt", "a"))
print("overall_errors = ", overall_errors, "\n", file = open("parallel_error_collision_P.txt", "a"))

Parallel(n_jobs=len(ns3_files))(delayed(start_process)(fil) for fil in ns3_files)

• You should provide some means for others to test the code, ideally on a simple input and on the problematic one. The code is certainly hard to read (too many nesting levels for a start...), which makes it harder to find the speed bottleneck. Have you tried profiling your code? Mar 31 '20 at 18:38
• @norok2 no I haven't tried profiling. Most of the mathematical operations that I use are linear, so it is confusing why it's so slow. I will try profiling.Also, do i need to upload some files where the program can be tested ? Mar 31 '20 at 18:41
• Uploading test files would be very helpful. Mar 31 '20 at 20:09
• @pacmaninbw I have added the testing files, link is in the question description. Apr 1 '20 at 0:20

I see a number of things that may help you improve your program.

Decompose the code into smaller functions

This code is very dense, very long and not well organized making it difficult to follow and understand. As a first step, I'd recommend extracting out smaller functions, such as to calculate error values. Each function should be small, well documented and testable.

Pass needed variables

Instead of relying on interspersed code and function declarations as in this code, gather things into a main function. Here's one way to do that:

if __name__ == "__main__":
prediction = 0 # 0 means no prediction and 1 is prediction enabled

ns3_files = ['EQ_100p_1Hz_50U.txt','EQ_100p_5Hz_50U.txt','EQ_100p_10Hz_50U.txt',
'EQ_100p_20Hz_50U.txt','EQ_100p_30Hz_50U.txt','EQ_100p_40Hz_50U.txt',
'EQ_100p_50Hz_50U.txt','EQ_100p_100Hz_50U.txt']

sumo_file = ['Loc_Circle_50U.txt']
sumo_df = pd.read_csv(sumo_file[0], delim_whitespace = True)
sumo_headers = {'Time', 'VID', 'Pos(X)', 'Pos(Y)', 'Vel(X)', 'Vel(Y)'}
print("sumo file ", sumo_file[0], " is OK")

[start_process(fil, sumo_df) for fil in ns3_files]


Write more "pythonic" code

Python makes extensive use of data structure such as lists, dictionaries and sets. If you find yourself writing long if constructs as in this code, STOP! There is almost always a better way to do it. For example, you'll see that in the sample code above there is a validate_headers function. Here's the definition

def validate_headers(df, headerset):
return headerset <= set(df.columns.values)


It uses the <= operator on two sets to determine whether one is a strict subset of the other. In this case, we're trying to assure that all of the required fields exist in a dataframe, so we pass the dataframe and the set of header names. Simple!

Don't do redundant work

The original code loads the entire dataframe of each file just to validate the header and then loads it again to actually process the data. This is pointless and wastes time and memory.

Use pandas as it is intended

Iterating over pandas data, row by row using an index is the very slowest possible way to process the data. Instead, you should seek to use vectorization. For example, if you want to create a new column errx for each row in the dataframe, you could write this:

df['errx'] = abs(df['TxTruePos(X)'] - df['TxHeadPos(X)'])


Using vectorized operations is the way pandas is intended to be used and is very efficient compared to using for loops. If you find you need still more performance, however, you can use numpy, which you're already including but not making much use of.

I haven't validated all of it, but it appears that much of the calculation may be redundant. For example, the program calculates an error_x_val but the ns3_data files appear to already have such a column. If that actually contains the data you need, use it instead of recalculating. If it doesn't, I'd suggest dropping it from the data frame if it's not useful. That can be done like this:
del df['Error(X)']

• I tried to implement most of the recommendations listed in your answer @Edward, but I guess my way of writing the program, in terms of the sequence of operations I follow and other things, is not optimum. Vectorization, as mentioned in the answer, was the most significant improvement bringer. Overall, the code has around 11000 pairs of vehicles and each pair needs to be processed, so maybe parallelizing the for loop that calls calc_error(c) could be the best thing I could do as each pair is independent, though not sure if that can be done. With the code as it is, this is the best I could do. Apr 6 '20 at 5:00