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.tkgraphic.tkplotevent import tkPlotEvent """ Split multiple data to multiple files """ #=================================== # 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() 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 = "--output_dir", type = "str", var_name = 'output_dir', opt_str = "--output_dir=path", desc = 'output dir', defval = None, optional = False) 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 split(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.output_dir}") 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'], 'XRF', '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_in = 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(): Cannot load [{cparams.output_plugin_dir}/{cparams.output_plugin_name}.py]", pause = True) print("Read input files:", cparams.infile) inf_in = filter_in.read_data(0, cparams.infile, app = app, cparams = cparams, is_print = True) sample = inf_in.get("sample_name", "") labels = inf_in["labels"] data_list = inf_in["data_list"] 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) nspectrum = inf_in["nspectrum"] for isp in range(nspectrum): inf = inf_in["meta"][isp] name = inf.get("name", "") outfile = os.path.join(cparams.output_dir, f"{isp+1:02d}-{sample}-{name}.txt") _inf = { "labels": [inf_in["labels"][isp]], "data_list": [inf_in["data_list"][isp]], } # print(f"Save to [{outfile}]") ret = cparams.out_modules[0].save_data(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(" Split multiple data to multiple files") print( "==========================================================================") # print(f"mode: {cparams.mode}") split(app, cparams) app.terminate(pause = True)#usage = usage) if __name__ == "__main__": main()