import os import sys import re import numpy as np from numpy import exp, log, log10, sqrt, sin, cos, tan from importlib import import_module from matplotlib import pyplot as plt from tklib.tkapplication import tkApplication from tklib.tkfile import tkFile from tklib.tkutils import getarg, getintarg, getfloatarg, pint, pfloat, split_file_path, replace_path, print_line from tklib.tkvariousdata import tkVariousData from tklib.tkfilter import tkFilter from tklib.tksci.tksci import Gaussian, Lorentzian, GaussLorentz from tklib.tksci.tkmatrix import make_matrix1, make_matrix2, make_matrix3 from tklib.tkcrystal.tkxrd import Xray_wavelengths from tklib.tkgraphic.tkplotevent import tkPlotEvent """ Extract data by specifying x- and y-ranges """ #=================================== # Global variables #=================================== markers = ['o', 's', '+', 'x', 'D', 'v', '^', '<', '>', '8', 'h', 'H'] colors = ['black', 'red', 'blue', 'darkgreen', 'darkorange', 'hotpink', 'lightgreen', 'cyan', 'yellow', 'magenta', 'chocolate', 'navy', 'slategray', 'olive' ] figsize = (12, 8) fontsize = 12 legend_fontsize = 8 #============================= # Treat argments #============================= def usage(app = None, cparams = None): cparams = app.get_params() # app.usage(infile = cparams.infile) for s in app.usage_str.split('\n'): cmd = 'print({})'.format(s.rstrip()) eval(cmd) def initialize(app = None, cparams = None): def_plugin_dir = os.path.join(os.environ["tkprog_X_path"], 'plugin', "filter") app.add_argument(opt = "--input_plugin_dir", type = "str", var_name = 'input_plugin_dir', opt_str = "--input_plugin_dir=dir", desc = 'input_plugin_dir', defval = def_plugin_dir, optional = False) app.add_argument(opt = "--input_plugin_name", type = "str", var_name = 'input_plugin_name', opt_str = "--input_plugin_name=file_body", desc = 'input_plugin_name', defval = 'xlsx2xlsx', optional = False) app.add_argument(opt = "--output_plugin_dir", type = "str", var_name = 'output_plugin_dir', opt_str = "--output_plugin_dir=dir", desc = 'output_plugin_dir', defval = def_plugin_dir, optional = False) app.add_argument(opt = "--output_plugin_name", type = "str", var_name = 'output_plugin_name', opt_str = "--output_plugin_name=file_body", desc = 'output_plugin_name', defval = 'xlsx2xlsx', optional = False) app.add_argument(opt = "--infile", type = "str", var_name = 'infile', opt_str = "--infile=path", desc = 'input file path', defval = None, optional = False) app.add_argument(opt = "--outfile", type = "str", var_name = 'outfile', opt_str = "--outfile=path", desc = 'output file path', defval = None, optional = False) app.add_argument(opt = "--xmin", type = "float", var_name = 'xmin', opt_str = "--xmin=val", desc = 'xmin to collect', defval = -1.0e100, optional = True) app.add_argument(opt = "--xmax", type = "float", var_name = 'xmax', opt_str = "--xmax=val", desc = 'xmax to collect', defval = 1.0e100, optional = True) app.add_argument(opt = "--ymin", type = "float", var_name = 'ymin', opt_str = "--ymin=val", desc = 'ymin to collect', defval = -1.0e100, optional = True) app.add_argument(opt = "--ymax", type = "float", var_name = 'ymax', opt_str = "--ymax=val", desc = 'ymax to collect', defval = 1.0e100, optional = True) app.add_argument(opt = "--nxmax", type = "int", var_name = 'nxmax', opt_str = "--nxmax=val", desc = 'nxmax to collect', defval = 1000000000, optional = True) def update_vars(app = None, cparams = None): args_opt, args_idx, args_vars = app.read_args(vars = cparams, check_allowed_args = True, apply_default = True) if args_opt is None: error_no = args_idx error_message = args_vars app.terminate("\n\n" + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!\n" + f"! {error_message}\n" + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!", usage = app.usage) def extract(app, cparams): print() print(f"input_plugin: {cparams.input_plugin_dir}/{cparams.input_plugin_name}") print(f"output_plugin: {cparams.output_plugin_dir}/{cparams.output_plugin_name}") print(f"input file : {cparams.infile}") print(f"output file: {cparams.outfile}") print(f"x range: {cparams.xmin} - {cparams.xmax}") print(f"y range: {cparams.ymin} - {cparams.ymax}") print(f"nxmax: {cparams.nxmax}") def read_file(path, app, cparams): module = None for i in range(len(modules)): name = module_names[i] m = modules[i] file_type = m.check_file_type(path, app = app, cparams = cparams) print(f"try [{name}] for [{path}]: file_type={file_type}") if file_type is not None and 'Error' not in file_type: module = m module_name = name break if module is None and 'Error': return None, None inf = module.read_data(path, app = app, cparams = cparams) return module_name, module, inf if cparams.input_plugin_name == '': cparams.plugin_dir = os.path.join(os.environ['tkprog_X_path'], 'XRD', 'filter') print() print(f"Load modules from [{cparams.plugin_dir}] (auto judge)") module_names, modules = app.load_modules(cparams.plugin_dir, "*.py", target = "read_data", is_print = True) for m in modules: input_type = m.get_input_type(app = app, cparams = cparams) output_type = m.get_output_type(app = app, cparams = cparams) print(f" {m.name}: input_type={input_type} output_type={output_type}") module_name, module_input, inf_input = read_file(cparams.infile, app, cparams) # print("module_name=", module_name) if len(module_name) == 0: app.terminate(f"Error in extract_data.extract(): Can not find a module for [{cparams.infile}]", pause = True) cparams.in_module_names = [module_name] cparams.in_modules = [module_input] inf_obs = inf_input else: print("") print(f"Load input module from {cparams.input_plugin_dir}/{cparams.input_plugin_name}.py") filter_in = tkFilter(app = app, cparams = cparams, plugin_dir = cparams.input_plugin_dir, module_file = f"{cparams.input_plugin_name}.py") cparams.in_module_names, cparams.in_modules = filter_in.load(target = "read_data", is_print = True) print("cparams.in_module_names=", cparams.in_module_names) if len(cparams.in_module_names) == 0: app.terminate(f"\nError in extract(): Can not load [{cparams.output_plugin_dir}/{cparams.output_plugin_name}.py]", pause = True) print("Read input files:", cparams.infile) inf_obs = filter_in.read_data(0, cparams.infile, app = app, cparams = cparams, is_print = True) # print("inf_obs[labels]=", inf_obs["labels"]) # print("data=", inf_obs["data_list"]) labels_in = inf_obs["labels"][0] data_in = inf_obs["data_list"][0] data_in = list(zip(*sorted(zip(*data_in)))) ncols_in = len(data_in) ndata_in = len(data_in[0]) nskip = int(ndata_in / 25) if nskip <= 0: nskip = 1 print() print("ndata_in=", ndata_in) for l in labels_in: print(f"{l:>10}\t", end = '') print() for i in range(0, ndata_in, nskip): for j in range(ncols_in): print(f"{data_in[j][i]:10.4g}\t", end = '') print() data_out0 = [] for i in range(ncols_in): data_out0.append([]) for i in range(ndata_in): x = data_in[0][i] y = data_in[1][i] if x < cparams.xmin or cparams.xmax < x: continue if y < cparams.ymin or cparams.ymax < y: continue for j in range(ncols_in): data_out0[j].append(data_in[j][i]) ndata_out0 = len(data_out0[0]) nskip = int(ndata_out0 / cparams.nxmax) if nskip <= 0: nskip = 1 print() print(f"ndata_out0={ndata_out0}") print(f"Input data are collected every {nskip} data to adjust to nxmax={cparams.nxmax}") data_out = [] for i in range(ncols_in): data_out.append([]) for i in range(0, ndata_out0, nskip): x = data_out0[0][i] y = data_out0[1][i] if x < cparams.xmin or cparams.xmax < x: continue if y < cparams.ymin or cparams.ymax < y: continue for j in range(ncols_in): data_out[j].append(data_out0[j][i]) ndata_out = len(data_out[0]) print("ndata_out=", ndata_out) for l in labels_in: print(f"{l:>10}\t", end = '') print() for i in range(0, ndata_out): for j in range(ncols_in): print(f"{data_out[j][i]:10.4g}\t", end = '') print() print() print(f"Load output module from {cparams.output_plugin_dir}/{cparams.output_plugin_name}.py") filter_out = tkFilter(app = app, cparams = cparams, plugin_dir = cparams.output_plugin_dir, module_file = f"{cparams.output_plugin_name}.py") cparams.out_module_names, cparams.out_modules = filter_out.load(target = "save_data", is_print = True) print("cparams.out_module_names=", cparams.out_module_names) if len(cparams.out_module_names) == 0: app.terminate(f"\nError in extract(): Can not load [{cparams.output_plugin_dir}/{cparams.output_plugin_name}.py]", pause = True) inf = { "labels": [labels_in], "data_list": [data_out], } print(f"Save to [{cparams.outfile}]") ret = cparams.out_modules[0].save_data(cparams.outfile, inf, app = app, cparams = cparams, is_print = True) if ret is None: app.terminate("\nError in extract(): Can not write to [{cparams.outfile}]", pause = True) def main(): #================================================================== # Initialize parameters #================================================================== app = tkApplication() cparams = app.get_params() logfile = app.replace_path(None, template = ["{dirname}", "{filebody}-out.txt"]) # logfile = app.replace_path(cparams.infile, template = ["{dirname}", "{filebody}-out.txt"]) print(f"Open logfile [{logfile}]") app.redirect(targets = ["stdout", logfile], mode = 'w') initialize(app, cparams) update_vars(app, cparams) print("") print( "==========================================================================") print(" Extract data by specifying x- and y-ranges") print( "==========================================================================") # print(f"mode: {cparams.mode}") extract(app, cparams) app.terminate(pause = True)#usage = usage) if __name__ == "__main__": main()