Added dynamic simulation for use with distributed systems and incremental simulations Added module based exports Added test for distributed systems [skip ci]
83 lines
3.4 KiB
Python
83 lines
3.4 KiB
Python
import argparse
|
|
import csv
|
|
import sys
|
|
|
|
|
|
def read_matrices_from_csv(file_path):
|
|
with open(file_path, newline='') as csvfile:
|
|
reader = csv.reader(csvfile, delimiter=' ', skipinitialspace=True)
|
|
matrices, matrix, line_numbers = [], [], []
|
|
line_number = 0
|
|
|
|
for row in reader:
|
|
line_number += 1
|
|
if row:
|
|
matrix.append([float(item) for item in row])
|
|
line_numbers.append(line_number)
|
|
else:
|
|
if matrix:
|
|
matrices.append((matrix, line_numbers))
|
|
matrix, line_numbers = [], []
|
|
|
|
if matrix: # Add the last matrix if there's no blank line at the end
|
|
matrices.append((matrix, line_numbers))
|
|
|
|
return matrices
|
|
|
|
def compare_matrices(matrix_a_tuple, matrix_b_tuple, tolerance=0.01):
|
|
data_a, line_numbers_a = matrix_a_tuple
|
|
data_b, line_numbers_b = matrix_b_tuple
|
|
max_diff = 0
|
|
|
|
if len(data_a) != len(data_b):
|
|
return False, f"Matrix size mismatch: Matrix A has {len(data_a)} rows, Matrix B has {len(data_b)} rows.", None, None, max_diff
|
|
|
|
for row_index, (row_a, row_b, line_num_a, line_num_b) in enumerate(zip(data_a, data_b, line_numbers_a, line_numbers_b)):
|
|
if len(row_a) != len(row_b):
|
|
return False, f"Row size mismatch at line {line_num_a} (File A) and line {line_num_b} (File B).", line_num_a, line_num_b, max_diff
|
|
|
|
for col_index, (elem_a, elem_b) in enumerate(zip(row_a, row_b)):
|
|
if abs(elem_a - elem_b) > max_diff:
|
|
max_diff = abs(elem_a - elem_b)
|
|
if abs(elem_a - elem_b) > tolerance:
|
|
return False, f"Mismatch at line {line_num_a}, column {col_index+1} (File A) vs line {line_num_b}, column {col_index+1} (File B):\nA({elem_a}) vs B({elem_b}).", line_num_a, line_num_b, max_diff
|
|
|
|
return True, "Matrices are equal.", None, None, max_diff
|
|
|
|
def compare_csv_files(file_path1, file_path2, tolerance=0):
|
|
matrices1 = read_matrices_from_csv(file_path1)
|
|
matrices2 = read_matrices_from_csv(file_path2)
|
|
max_difference = 0
|
|
|
|
if len(matrices1) != len(matrices2):
|
|
return False, "Number of matrices in files do not match.", None, None, max_difference
|
|
|
|
for index, (matrix_a_tuple, matrix_b_tuple) in enumerate(zip(matrices1, matrices2)):
|
|
equal, message, line_num_a, line_num_b, max_diff = compare_matrices(matrix_a_tuple, matrix_b_tuple, tolerance)
|
|
if max_diff > max_difference:
|
|
max_difference = max_diff
|
|
if not equal:
|
|
return False, f"In Matrix pair {index}: {message}", line_num_a, line_num_b, max_difference
|
|
|
|
return True, "All matrices are equal.", None, None, max_difference
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Compare two CSV files containing matrices.')
|
|
parser.add_argument('file_a', help='Path to File A')
|
|
parser.add_argument('file_b', help='Path to File B')
|
|
parser.add_argument('--tolerance', type=float, default=0, help='Tolerance for comparison (default: 0)')
|
|
parser.add_argument('--silent', action='store_true', help='Run in silent mode without printing details')
|
|
args = parser.parse_args()
|
|
|
|
are_equal, message, line_num_a, line_num_b, max_difference = compare_csv_files(args.file_a, args.file_b, args.tolerance)
|
|
|
|
if not args.silent:
|
|
print(message)
|
|
|
|
if not are_equal:
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|