Keep single file data into memory (2c842681) · Commits · Giacomo Mulas / NP_TMcode

src/scripts/filter_constants.py

+158 −65

Original line number	Diff line number	Diff line
		@@ -61,7 +61,33 @@ def main():
		else:
		split_input = config['input_files'].split(',')
		if (len(split_input) == 1):
		result = scan_single_file(config)
		scan_info = scan_single_file(config, split_input[0])
		result = scan_info['exit_code']
		#breakpoint()
		if (result == 0):
		write_output_file(config['output_files'], scan_info)
		# INFO section
		num_filtered_data = scan_info['num_filtered_data']
		num_orig_data = scan_info['num_orig_data']
		msg_filtered = "{0:d} filtered lines".format(num_filtered_data) if num_filtered_data != 1 else "1 filtered line"
		msg_orig = "{0:d} input lines".format(num_orig_data) if num_orig_data != 1 else "1 input line"
		max_distance = get_max_distance(scan_info)
		msg_distance1 = "INFO: maximum absolute distance was {0:.5g} at {1:.5e} {2:s} in the {3:s} part".format(
		max_distance['max_difference'],
		max_distance['max_wavelength'],
		config['wl_units'],
		max_distance['differing_set']
		)
		msg_distance2 = " (fitted value is {0:.5g}, actual data value is {1:.5g}).".format(
		max_distance['max_fitted'],
		max_distance['max_value']
		)
		print("INFO: extracted %s out of %s."%(msg_filtered, msg_orig))
		print(msg_distance1)
		print(msg_distance2)
		if (config['make_plots']):
		plot_data(scan_info, config['wl_units'])
		# end result == 0 check
		else:
		result = scan_multiple_files(config)
		return result
		@@ -73,18 +99,19 @@ def main():
		# distribution of the input data and a segmented linear interpolation touching all
		# the filtered data.
		#
		# \param[in] wl_orig: `array-like` Array of the input data wavelengths.
		# \param[in] reps_orig: `array-like` Array of the real part of the input optical functions.
		# \param[in] ieps_orig: `array-like` Array of the imaginary part of the input optical functions.
		# \param[in] wl_filtered: `array-like` Array of the filtered data wavelengths.
		# \param[in] reps_filtered: `array-like` Array of the real part of the filtered optical functions.
		# \param[in] ieps_filtered: `array-like` Array of the imaginary part of the filtered optical functions.
		# \param[in] scan_info: `dict` A dictionary containing the results of a file scan.
		# \returns result: `dict` A dictionary containing the wavelength of the maximum difference
		# (`max_wavelength`), the value of the input data at that wavelength (`mav_value`),
		# the value of the interpolated filtered functions at the same wavelength (`max_fitted`),
		# the offset between interpolation and data (`max_difference`), and the set of values
		# where the difference was observed (`differing_set`, being either REAL or IMAGINARY).
		def get_max_distance(wl_orig, reps_orig, ieps_orig, wl_filtered, reps_filtered, ieps_filtered):
		def get_max_distance(scan_info):
		wl_orig = scan_info['wl_orig']
		reps_orig = scan_info['reps_orig']
		ieps_orig = scan_info['ieps_orig']
		wl_filtered = scan_info['wl_filtered']
		reps_filtered = scan_info['reps_filtered']
		ieps_filtered = scan_info['ieps_filtered']
		max_difference = 0.0
		max_fitted = 0.0
		max_value = 0.0
		@@ -202,8 +229,49 @@ def parse_arguments():
		else:
		raise Exception("Unrecognized argument \"{0:s}\"!".format(arg))
		# end for loop
		if (config['output_files'] == ''):
		split_in = config['input_files'].split(',')
		for name_in in split_in:
		name_out = name_in.split('.')[0] + "_filtered.csv"
		if (config['output_files'] == ''):
		config['output_files'] = name_out
		else:
		config['output_files'] += ",{0:s}".format(name_out)
		else:
		split_in = config['input_files'].split(',')
		split_out = config['output_files'].split(',')
		if (len(split_in) != len(split_out)):
		raise Exception("Output list does not match input!")
		for ni in range(len(split_in)):
		input_name = split_in[ni]
		output_name = split_out[ni]
		if (input_name == output_name):
		raise Exception("No input file overwriting allowed!")
		return config

		## \brief Make a quick-look plot with MATPLOTLIB.
		#
		# \param[in] scan_info: `dict` A dictionary with the scanned data file.
		# \param[in] wl_units: `string` The name of the units for the wavelength scale.
		def plot_data(scan_info, wl_units):
		# Plot making section
		wl_orig = scan_info['wl_orig']
		reps_orig = scan_info['reps_orig']
		ieps_orig = scan_info['ieps_orig']
		wl_filtered = scan_info['wl_filtered']
		reps_filtered = scan_info['reps_filtered']
		ieps_filtered = scan_info['ieps_filtered']
		plt.plot(wl_orig, reps_orig, color='red', marker='', ls='-', label=r"Original $\mathfrak{Re}(\varepsilon)$")
		plt.plot(wl_filtered, reps_filtered, color='red', marker='o', ls='', label=r"Filtered $\mathfrak{Re}(\varepsilon)$")
		plt.plot(wl_orig, ieps_orig, color='blue', marker='', ls='-', label=r"Original $\mathfrak{Im}(\varepsilon)$")
		plt.plot(wl_filtered, ieps_filtered, color='blue', marker='o', ls='', label=r"Filtered $\mathfrak{Im}(\varepsilon)$")
		plt.xlabel("Wavelength ({0:s})".format(wl_units))
		plt.ylabel(r"$\mathfrak{Re}(\varepsilon)$\|$\mathfrak{Im}(\varepsilon)$")
		plt.legend(loc="best")
		plt.show()
		# end plot making section


		## \brief Print a command-line help summary.
		def print_help():
		print(" ############################################### ")
		@@ -249,24 +317,35 @@ def print_help():
		# is shown as a plot, if MATPLOTLIB is available on the system.
		#
		# \param[in] config: `dict` A dictionary containing the script configuration.
		# \return result: `int` An integer exit code (0 if successful).
		def scan_single_file(config):
		result = 0
		# \param[in] file_name: `string` The name of the single input file.
		# \return result: `dict` A dictionary containing the results of the scan,
		# including "exit_code" (`int`, 0 if succesful), "wl_orig" (`array-like`, the
		# original wavelength scale), "reps_orig" (`array-like`, the original real
		# parts of the dielectric functions), "ieps_orig" (`array-like`, the original
		# imaginary parts of the dielectric functions), "wl_filtered" (`array-like`,
		# the filtered wavelength scale), "reps_filtered" (`array-like`, the filtered
		# real parts of the dielectric functions), and "ieps_filtered" (`array-like`,
		# the filtered imaginary parts of the dielectric functions).
		def scan_single_file(config, file_name):
		result = {
		'exit_code': -1,
		'header': "",
		'num_read_lines': 0,
		'num_orig_data': 0,
		'num_filtered_data': 0,
		'wl_orig': [],
		'reps_orig': [],
		'ieps_orig': [],
		'wl_filtered': [],
		'reps_filtered': [],
		'ieps_filtered': []
		}
		try:
		file_name = config['input_files']
		input_file = open(file_name, 'r')
		file_line = input_file.readline()
		num_read_lines = 1
		num_orig_data = 0
		num_filtered_data = 0
		output_name = config['output_files']
		if (output_name == ''):
		output_name = file_name.split('.')[0] + "_filtered.csv"
		if (output_name == config['input_files']):
		print("ERROR: overwriting of input files is not supported!")
		result = 2
		return result
		output_file = open(output_name, 'w')
		wl_factor = 1.0e6
		wl_units = config['wl_units']
		if (wl_units in ["nanometers", "nm"]):
		@@ -279,12 +358,12 @@ def scan_single_file(config):
		wl_factor = 1.0e1
		elif (wl_units in ["meters", "m"]):
		wl_factor = 1.0
		wl_orig = []
		reps_orig = []
		ieps_orig = []
		wl_filtered = []
		reps_filtered = []
		ieps_filtered = []
		wl_orig = result['wl_orig']
		reps_orig = result['reps_orig']
		ieps_orig = result['ieps_orig']
		wl_filtered = result['wl_filtered']
		reps_filtered = result['reps_filtered']
		ieps_filtered = result['ieps_filtered']
		step = config['step']
		threshold = config['threshold']
		wl0 = 0.0
		@@ -297,7 +376,7 @@ def scan_single_file(config):
		last_dieps = 0.0
		while (file_line != ""):
		if (file_line.startswith('#')):
		output_file.write(file_line)
		result['header'] += file_line
		file_line = input_file.readline()
		num_read_lines += 1
		continue
		@@ -314,7 +393,6 @@ def scan_single_file(config):
		ieps_orig.append(ieps0)
		num_orig_data += 1
		if (wl0 >= config['wl_start']):
		output_file.write(file_line)
		wl_filtered.append(wl0 * wl_factor)
		reps_filtered.append(reps0)
		ieps_filtered.append(ieps0)
		@@ -352,7 +430,6 @@ def scan_single_file(config):
		reps = ry0 + dry * dx / (x1 - x0)
		ieps = iy0 + diy * dx / (x1 - x0)
		# write a line if step is enabled and satisfied
		output_file.write(file_line)
		wl_filtered.append(wl * wl_factor)
		reps_filtered.append(reps)
		ieps_filtered.append(ieps)
		@@ -367,7 +444,6 @@ def scan_single_file(config):
		ipeak = (dieps * last_dieps < 0.0)
		if ((rpeak or ipeak) and can_write):
		# write a line if peaks are enabled and satisfied
		output_file.write(file_line)
		can_write = False
		wl_filtered.append(wl1 * wl_factor)
		reps_filtered.append(reps1)
		@@ -385,7 +461,6 @@ def scan_single_file(config):
		rel_dreps *= -1.0
		if ((rel_dreps > 1.0 + threshold or rel_dreps < 1.0 - threshold) and can_write):
		# write a line if tolerance is violated
		output_file.write(file_line)
		can_write = False
		wl_filtered.append(wl1 * wl_factor)
		reps_filtered.append(reps1)
		@@ -401,7 +476,6 @@ def scan_single_file(config):
		rel_dieps *= -1.0
		if ((rel_dieps > 1.0 + threshold or rel_dieps < 1.0 - threshold) and can_write):
		# write a line if tolerance is violated
		output_file.write(file_line)
		can_write = False
		wl_filtered.append(wl1 * wl_factor)
		reps_filtered.append(reps1)
		@@ -414,51 +488,70 @@ def scan_single_file(config):
		# end of wl0 == 0.0 check
		else:
		print("ERROR: invalid input file %s at line %d!"%(file_name, num_read_lines))
		result = 1
		result['exit_code'] = 1
		break # while loop
		# end of len(split_line) check
		file_line = input_file.readline()
		num_read_lines += 1
		# end of while loop
		input_file.close()
		output_file.close()
		# INFO section
		msg_filtered = "{0:d} filtered lines".format(num_filtered_data) if num_filtered_data != 1 else "1 filtered line"
		msg_orig = "{0:d} input lines".format(num_orig_data) if num_orig_data != 1 else "1 input line"
		max_distance = get_max_distance(wl_orig, reps_orig, ieps_orig, wl_filtered, reps_filtered, ieps_filtered)
		msg_distance1 = "INFO: maximum absolute distance was {0:.5g} at {1:.5e} {2:s} in the {3:s} part".format(
		max_distance['max_difference'],
		max_distance['max_wavelength'],
		config['wl_units'],
		max_distance['differing_set']
		)
		msg_distance2 = " (fitted value is {0:.5g}, actual data value is {1:.5g}).".format(
		max_distance['max_fitted'],
		max_distance['max_value']
		)
		print("INFO: extracted %s out of %s."%(msg_filtered, msg_orig))
		print(msg_distance1)
		print(msg_distance2)
		# Plot making section
		if (config['make_plots']):
		plt.plot(wl_orig, reps_orig, color='red', marker='', ls='-', label=r"Original $\mathfrak{Re}(\varepsilon)$")
		plt.plot(wl_filtered, reps_filtered, color='red', marker='o', ls='', label=r"Filtered $\mathfrak{Re}(\varepsilon)$")
		plt.plot(wl_orig, ieps_orig, color='blue', marker='', ls='-', label=r"Original $\mathfrak{Im}(\varepsilon)$")
		plt.plot(wl_filtered, ieps_filtered, color='blue', marker='o', ls='', label=r"Filtered $\mathfrak{Im}(\varepsilon)$")
		plt.xlabel("Wavelength ({0:s})".format(config['wl_units']))
		plt.ylabel(r"$\mathfrak{Re}(\varepsilon)$\|$\mathfrak{Im}(\varepsilon)$")
		plt.legend(loc="best")
		plt.show()
		# end plot making section
		if (result['exit_code'] < 0):
		result['exit_code'] = 0
		result['num_read_lines'] = num_read_lines
		result['num_orig_data'] = num_orig_data
		result['num_filtered_data'] = num_filtered_data
		except FileNotFoundError as ex:
		print("ERROR: file not found %s!"%config['input_files'])
		result = 1
		result['exit_code'] = 1
		return result

		## \brief Filter multiple files based on the configuration options.
		#
		# A sequence of optical function data files is filtered according to the
		# same samplig grid, resulting in a set of files ready for use in the same
		# simulation.
		#
		# \param[in] config: `dict` A dictionary containing the script configuration.
		# \return result: `dict` A dictionary containing the results of the scan,
		# including "exit_code" (`int`, 0 if succesful), "wl_orig" (`array-like`, the
		# original wavelength scale), "reps_orig" (`array-like`, the original real
		# parts of the dielectric functions), "ieps_orig" (`array-like`, the original
		# imaginary parts of the dielectric functions), "wl_filtered" (`array-like`,
		# the filtered wavelength scale), "reps_filtered" (`array-like`, the filtered
		# real parts of the dielectric functions), and "ieps_filtered" (`array-like`,
		# the filtered imaginary parts of the dielectric functions).
		def scan_multiple_files(config):
		result = 0
		result = {
		'exit_code': -1,
		'header': "",
		'num_read_lines': 0,
		'num_orig_data': 0,
		'num_filtered_data': 0,
		'wl_orig': [],
		'reps_orig': [],
		'ieps_orig': [],
		'wl_filtered': [],
		'reps_filtered': [],
		'ieps_filtered': []
		}
		return result

		## \brief Write the filtered data to an output file.
		#
		# \param[in] file_name: `string` The name of the file to be written.
		# \param[in] scan_info: `dict` A dictionary with the scanned data file.
		def write_output_file(file_name, scan_info):
		output_file = open(file_name, 'w')
		output_file.write(scan_info['header'])
		for i in range(len(scan_info['wl_filtered'])):
		file_line = "{0:.5E},{1:.5E},{2:.5E}\n".format(
		scan_info['wl_filtered'][i],
		scan_info['reps_filtered'][i],
		scan_info['ieps_filtered'][i]
		)
		output_file.write(file_line)
		output_file.close()

		## \brief Exit code (0 for success).
		exit_code = main()
		exit(exit_code)