#!/usr/bin/env python3 import csv import os import sys import subprocess import matplotlib.pyplot as plt import matplotlib.patches as patches import numpy as np from subprocess import PIPE verbose = True # New things added to make gifs --------------- def create_gif_from_images(input_pattern, output_filename, framerate=10): """ Creates a GIF from a sequence of images using FFmpeg. Args: input_pattern (str): The pattern for input image files (e.g., "image%03d.png"). output_filename (str): The name of the output GIF file (e.g., "output.gif"). framerate (int): The desired frame rate for the GIF (frames per second). """ # first create the pallete command = [ "ffmpeg", "-y", "-i", input_pattern, "-vf", "palettegen", "palette.png", ] try: subprocess.run(command, check=True) print(f"pallete '{output_filename}' created successfully.") except subprocess.CalledProcessError as e: print(f"Error creating pallet: {e}") command = [ "ffmpeg", "-y", "-framerate", str(framerate), "-i", input_pattern, "-i", "palette.png", "-lavfi", "paletteuse", output_filename ] try: subprocess.run(command, check=True) print(f"GIF '{output_filename}' created successfully.") except subprocess.CalledProcessError as e: print(f"Error creating GIF: {e}") def create_frame(start_time, stop_time, index, vehicle_log_data, lims, frame_dir="frames"): """ Creates a frame at time t. Args: time (float): the time to plot. index (int) : the index to assign to this frame vehicle_log_data: dict of vehicle log info. example vehicle_log_data['abe']['x'] is all the x values. expected fields are x,y,and time lims : the plot limits. frame_dir : directory to save the file in """ if not os.path.exists(frame_dir): os.makedirs(frame_dir) fig, ax = plt.subplots(figsize=(6, 6)) # plot the data from each vehicle for legend_name, this_vehicle in vehicle_log_data.items(): # get the index of times # inialize to the last index i_t_start = len(this_vehicle['t'])-1 i_t_stop = len(this_vehicle['t'])-1 got_start = False got_stop = False for rec_index, rec_time in enumerate(this_vehicle['t']): if (rec_time > start_time) and not got_start: i_t_start = rec_index got_start = True if (rec_time > stop_time) and not got_stop: i_t_stop = rec_index got_stop = True if (got_start and got_stop): break # plot trace from start to stop ax.plot(this_vehicle['x'][i_t_start:i_t_stop], this_vehicle['y'][i_t_start:i_t_stop], label=legend_name) # plot the vehicle position at this time triangle_size = 4 triangle_vertices = np.array([[triangle_size*1.5, 0.0], [0.0, -triangle_size/2.0], [0.0, triangle_size/2.0]]) triangle_vertices = np.transpose(triangle_vertices) # rotate triangle theta = (90 - this_vehicle['hdg'][i_t_stop]) * 3.1415/180.0 rot_matrix = np.array([[np.cos(theta), -np.sin(theta)], [np.sin(theta), np.cos(theta)]]) tri_rotated = np.transpose(np.matmul(rot_matrix, triangle_vertices)) # translate to the vehicle position veh_x = this_vehicle['x'][i_t_stop] veh_y = this_vehicle['y'][i_t_stop] trans_mat = np.array([[veh_x, veh_y], [veh_x, veh_y], [veh_x, veh_y]]) veh_symbol = tri_rotated + trans_mat triangle_patch = patches.Polygon(veh_symbol, closed=True, edgecolor='black', facecolor='white', linewidth=2, alpha=0.7) ax.add_patch(triangle_patch) #ax.plot(this_vehicle['x'][i_t_stop], this_vehicle['y'][i_t_stop], 'k.', ms=15) ax.set_xlim(lims['min_x'] * 1.1, lims['max_x'] * 1.1) ax.set_ylim(lims['min_y'] * 1.1, lims['max_y'] * 1.1) time_int = int(stop_time) ax.set_title(f"time {time_int}s") #plt.axis('equal') ax.set_aspect('equal') plt.xlabel("x (m)") plt.ylabel("y (m)") plt.grid(which='both', linewidth=1.0) #plt.legend(loc='lower left') filename = os.path.join(frame_dir, f"frame_{index:03d}.png") plt.savefig(filename) #, transparent=False) plt.close(fig) # Close the figure to free up memory return filename #-------------- def vprint(string): """ The vprint function is a wrapper for the print function that only prints if the verbose flag is set. :param string: Print a string if the verbose flag is set :return: None :doc-author: Trelent """ if (verbose): print(string) def populate_xy(input_file): """ The populate_xy function takes in a csv file and returns a list of x values, y values, and timestamps. :param input_file: Specify the file that we want to read from :return: A list of x values, y values, and timestamps :doc-author: Trelent """ x = [] y = [] hdg = [] t = [] with open(input_file, 'r', encoding="utf-8") as file: reader = csv.reader(file, delimiter=' ') # next(reader) # skip header row for row in reader: timestamp, x_value, y_value, hdg_value = row x.append(float(x_value)) y.append(float(y_value)) hdg.append(float(hdg_value)) t.append(float(timestamp)) return x, y, hdg, t def remove_extension(fname): """ The remove_extension function takes a file name as an argument and returns the same file name with the extension removed. :param fname: Specify the file name that we want to remove the extension from :return: A string without the file extension :doc-author: Trelent """ last_dot_index = fname.rfind(".") if last_dot_index != -1: fname = fname[:last_dot_index] return fname def prepare_alog(input_file): """ The prepare_alog function takes in an alog file and runs the aloggrep command on it to process it into a csv file. It then returns the name of this new csv file. If the input is already a csv, then prepare_alog just returns that same file. :param input_file: Specify the alog file to process :return: The input file if it is a csv file and returns the output of aloggrep if it is an alog file :doc-author: Trelent """ # Determines if a file is an alog file or a csv file (checks the header) def is_alog(fname): """Determines if a file is an alog file (checks the header for a %) Args: fname (string): file name Returns: bool: if the file is an alog, return true """ return open(fname, 'r', encoding="utf-8").readline().startswith('%') # If the file is an alog file, run aloggrep to process it into a csv file if is_alog(input_file): output_file = remove_extension(input_file) + "_data.alog" output_file_real = remove_extension(input_file) + "_data.csv" # Remove the output file if it already exists to prevent aloggrep from asking to overwrite # subprocess.run(f"rm {output_file}", shell=True) script = "aloggrep "+input_file+" NODE_REPORT_LOCAL "+\ output_file + " -sd --format=time:val --csw --subpat=x:y:hdg " subprocess.run(script, shell=True, check=True) subprocess.run( f"mv {output_file} {output_file_real}", shell=True, check=True) return output_file_real # If the file is a csv file, just return it return input_file def find_alogs(): """ The find_alogs function searches the current directory for all alog files. It then returns a list of strings containing the full path to each alog file. :return: A list of alog files in the current directory :doc-author: Trelent """ alog_files = [] path = os.getcwd() for root, dirs, files in os.walk(path): # ignore subdirectories that begin with a period dirs[:] = [d for d in dirs if not d.startswith('.')] # ignore these subdirectories as well dirs[:] = [d for d in dirs if not d == 'src'] dirs[:] = [d for d in dirs if not d == 'bin'] dirs[:] = [d for d in dirs if not d == 'build'] dirs[:] = [d for d in dirs if not d == 'lib'] dirs[:] = [d for d in dirs if not d == 'scripts'] for file in files: if (file.endswith(".alog") and not file.__contains__("SHORESIDE")): alog_files.append(os.path.join(root, file)) return alog_files def alog_vname(alog_file): """ The alog_vname function returns the vehicle name from an alog file. :param alog_file: Specify the alog file to be used :return: The vehicle name from an alog file :doc-author: Trelent """ script = "aloggrep "+alog_file + \ " NODE_REPORT_LOCAL --v --final --format=val --subpat=name" vname = subprocess.run( script, shell=True, stdout=PIPE, stderr=PIPE, check=True).stdout.decode('utf-8') assert type(vname) == str, "Error: subprocess.run returns non-string type" assert not (vname.__contains__(" exiting") ), f"Error: {script} exitied with error: {vname}" return vname def alog_mhash(alog_file): """Returns mission hash from alog Args: alog_file (string): input file name Returns: string: mission hash """ """Returns the mission hash""" script = "aloggrep "+alog_file + \ " MISSION_HASH --v --final --format=val --subpat=mhash" hash = subprocess.run(script, shell=True, stdout=PIPE, stderr=PIPE, check=True).stdout.decode('utf-8').strip() assert not (hash.__contains__(" exiting") ), f"Error: {script} exitied with error: {hash}" assert isinstance( hash, str), "Error: subprocess.run returns non-string type" return hash def extract_files(): """Finds all algo files (or uses the ones provided) Returns: list of strings, string: paths to each file, figure name """ files = [] figname = "figure.png" for (i, arg) in enumerate(sys.argv): # skip over the name of this script if (i == 0): continue # save the figure output name if (arg.__contains__("figname=")): figname = arg[len("figname="):] continue files.append(arg) if (len(files) == 0): files = find_alogs() return files, figname def handle_no_alogs(to_find_alogs): """ The handle_no_alogs function is used to find alog files in the current directory. If no alogs are found, it will print an error message and exit with a status of 1. Otherwise, it will return a list of all the alog files found. :param to_find_alogs: Determine if the user wants to find alogs or not :return: A list of alog files :doc-author: Trelent """ alog_files = [] if (to_find_alogs): alog_files = find_alogs() vprint("Found alogs: ") for alog in alog_files: vprint("\t"+alog) # no alogs specified or found if (len(alog_files) == 0): if (to_find_alogs): print("Error: no alog files found. Use -h or --help for usage.") else: print("Error: no alog files specified. Use -h or --help for usage.") exit(1) return alog_files def plot_alogs(alogs, figname, ignore_hash, frames, framesPerSec): """ The plot_alogs function takes in a list of alog files and plots them. It will also save the plot as a file with the name specified by figname. If no figure name is given, it will use the mission hash (if all alogs have same hash) or "figure" if not. The function can take in multiple arguments for alogs, but they must be separated by spaces. :param alogs: Pass in the alog files to be plotted :param figname: Set the name of the figure :param ignore_hash: Ignore the hash of the alog files :param file_type: Determine the file type of the output figure :return: A tuple of the mission hash and figure name :doc-author: Trelent """ mhash = "" frame_dir = 'frames' file_type = "gif" # Step 1. get all the run data from each log file and process # the vehicle state. Also make note of the hash and # x and y limits for later plotting. legends = set() veh_log_data = {} lims = {} lims['min_x'] = 0.0 lims['max_x'] = 0.0 lims['min_y'] = 0.0 lims['max_y'] = 0.0 lims['max_t'] = 0.0 for arg in alogs: # converts to csv if necessary alog_file = prepare_alog(arg) legend_name = alog_vname(arg) x, y, hdg, t = populate_xy(alog_file) subprocess.run(f"rm {alog_file}", shell=True, check=True) if (len(x) == 0) or (len(y) == 0): continue # Handles the case where the vehicle name is empty # Will attempt to assign incrementing numbers to each vehicle if (len(legend_name) == 0): legend_name = "0" if (legend_name in legends): # if it ends in a number, increment it. Otherwise, add a 2 to the end if (legend_name[-1].isdigit()): legend_name = legend_name[:-1] + str(int(legend_name[-1])+1) else: legend_name += "2" legends.add(legend_name) # poplulate this entry in the veh log data dict this_veh_data = {} this_veh_data['x'] = x this_veh_data['y'] = y this_veh_data['hdg'] = hdg this_veh_data['t'] = t veh_log_data[legend_name] = this_veh_data # update the lims for plotting if min(x) < lims['min_x']: lims['min_x'] = min(x) if max(x) > lims['max_x']: lims['max_x'] = max(x) if min(y) < lims['min_y']: lims['min_y'] = min(y) if max(y) > lims['max_y']: lims['max_y'] = max(y) if max(t) > lims['max_t']: lims['max_t'] = max(t) # get mhash if (mhash == ""): mhash = alog_mhash(arg) else: if not (ignore_hash or mhash == alog_mhash(arg)): print("Error: alog files have different mission hashes: \n\t" + mhash+"\t"+alog_mhash(arg)) print(" Use -i or --ignore-hash to ignore this error") exit(1) # if there is no figure name, use the mission hash if (figname == ""): figname = mhash+"."+file_type print("Using hash as figure name: "+figname) # Step 2. generate all the frames interval = lims['max_t'] / float(frames) filenames = [] for i in range(frames): # set the times to show the vehicle trail start_time = 0.0 if (i > 0): start_time = float(i-1) * interval stop_time = float(i) * interval filename = create_frame(start_time, stop_time, i, veh_log_data, lims, frame_dir="frames") vprint('Created frame: ' + filename) filenames.append(filename) # step 3. generate the gif files from the frames # ensure some figure name is set (no hash and no argument for figurename) if (figname == ""): figname = "figure."+file_type input_pattern = os.path.join(frame_dir, "frame_%03d.png") create_gif_from_images(input_pattern, figname, framesPerSec) # clean up filenames.append("palette.png") for file_path in filenames: # Check if the file exists before attempting to delete if os.path.exists(file_path): try: os.remove(file_path) #print(f"Successfully deleted: {file_path}") except OSError as e: vprint(f"Error deleting {file_path}: {e}") else: vprint(f"File not found, skipping: {file_path}") def display_help(): """Prints out all help info. """ print("Usage: alog2gif.py --fname=figure [OPTIONS] [file1] [file2] ... [fileN] ") print(" Adapted from alog2fimage.py written by Kevin Becker 2024 ") print(" ") print(" This python script will input several alog files, read the local ") print(" node reports of each alog file, and create a gif of the x and y ") print(" values of each alog file on a graph. If no files are specified, ") print(" it will pull all alog files in the current directory and all ") print(" subdirectories. ") print(" ") print(" --ignorehash -i Ignores if the mission hash is different ") print(" between alog files. ") print(" ") print(" --auto -a Finds all alogs in the current directory ") print(" and all subdirectories. ") print(" ") print(" --fname= The name of the output gif. If not ") print(" specified, it will attempt to find the last") print(" mission hash. If no hash is given, the ") print(" default is figure. ") print(" ") print(" --frames= The number of frames in the gif. The ") print(" mission time will be split into the ") print(" specificed number of frames. Default is 50") print(" ") print(" --fps= The number of frames per second in the gif ") print(" Default is 3 ") print(" ") print(" Examples: ") print(" alog2gif.py -a -i --fps=10 Find all the log files in the current ") print(" directory and make a gif at 10 fps ") print(" ") print(" alog2gif.py alog1.alog --frames=200 ") print(" Create a gif from alog1.alog with 200 ") print(" frames. ") print(" ") def main(): """ The main function is the entry point of this program. It takes in command line arguments and plots the data from alog files. :return: Nothing :doc-author: Trelent """ arg = [] alog_files = [] figname = "" frames = 50 framesPerSec = 3 ignore_hash = False to_find_alogs = False for (i, arg) in enumerate(sys.argv): # skip over the name of this script if (i == 0): continue if (arg.startswith("-h") or arg.startswith("--help")): display_help() exit(0) if (arg.startswith("--fname=")): figname = arg[len("--fname="):] vprint("\tUsing figure name: "+figname) continue if (arg.startswith("--ignorehash") or arg.startswith("-i")): ignore_hash = True vprint("\tIgnoring mhash") continue if (arg.startswith("--auto") or arg.startswith("-a")): to_find_alogs = True vprint("\tFind alogs...") continue if (arg.endswith(".alog") or arg.endswith(".csv")): alog_files.append(arg) vprint("\tGiven alog file: "+arg) continue if (arg.startswith("--frames")): frames = int(arg[len("--frames="):]) vprint("\tFrames: "+ str(frames)) continue if (arg.startswith("--fps")): framesPerSec = int(arg[len("--fps="):]) vprint("\tFrames Per Second: "+ str(framesPerSec)) continue assert False, "alog2image.py error: " + arg + \ " is not a valid argument. Use -h or --help for usage." if len(alog_files) == 0: alog_files = handle_no_alogs(to_find_alogs) plot_alogs(alog_files, figname, ignore_hash, frames, framesPerSec) if __name__ == '__main__': """Main function """ main()