4
\$\begingroup\$

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"]
#             print("slots=", slots, "current_time=",current_time,"sender=",sender,"receiver=",receiver)
            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)"])
            x_header=(current_bsm["TxHeadPos(X)"])
            y_header=(current_bsm["TxHeadPos(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))
#             print("x_actual",x_actual,"y_actual",y_actual,"x_header",x_header,"y_header",y_header,"error_x_val",error_x_val,"error_y_val",error_y_val) #OK
            mask_2 = (df_actual["Time"]==current_time)
            sender_velocity_x = (df_actual[mask_2])["Vel(X)"].values[0]
            sender_velocity_y = (df_actual[mask_2])["Vel(Y)"].values[0]
            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
                df_row = df_actual[mask_3] 
#                 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)
                receiver_mask = (sumo_df["VID"]==receiver) & (sumo_df["Time"]==(current_time + (j+1)))
                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)))
        sender_x_vel = (sumo_df[sender_mask])["Vel(X)"].values[0]
        sender_y_vel = (sumo_df[sender_mask])["Vel(Y)"].values[0]
        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

    receivers = b['RxId'].unique()
    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)):
        receiver = receivers[i] # for every receiver
        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 = b[mask]
            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)
\$\endgroup\$
4
  • 1
    \$\begingroup\$ 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? \$\endgroup\$ – norok2 Mar 31 '20 at 18:38
  • \$\begingroup\$ @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 ? \$\endgroup\$ – user3656142 Mar 31 '20 at 18:41
  • \$\begingroup\$ Uploading test files would be very helpful. \$\endgroup\$ – pacmaninbw Mar 31 '20 at 20:09
  • \$\begingroup\$ @pacmaninbw I have added the testing files, link is in the question description. \$\endgroup\$ – user3656142 Apr 1 '20 at 0:20
3
\$\begingroup\$

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)'} 
    if validate_headers(sumo_df, sumo_headers):
        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.

Understand your data

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)']
\$\endgroup\$
1
  • \$\begingroup\$ 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. \$\endgroup\$ – user3656142 Apr 6 '20 at 5:00

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.