Source code for patato.convenience_scripts.data_analysis.analyse_dce

#  Copyright (c) Thomas Else 2023.
#  License: MIT

import argparse
import glob
from os.path import join

from ...io.msot_data import PAData
from ...unmixing.unmixer import DCEAnalyser
from ...utils import sort_key
from ...utils.pipeline import run_pipeline




def init_argparse():
    parser = argparse.ArgumentParser(description="Analyse Gas Challenge Data.")
    parser.add_argument('input', type=str, help="Data Folder")
    parser.add_argument('-p', '--prefix', type=str, help="Gas Challenge name prefix")
    parser.add_argument('-w', '--window', type=int, help="Smoothing window size",
                        default=10)
    parser.add_argument('-b', '--buffer', type=int, help="Buffer around changes",
                        default=5)
    parser.add_argument('-d', '--display', type=bool, help="Display steps",
                        default=True)
    parser.add_argument('-s', '--sigma', type=float, help="Smoothing window sigma",
                        default=4)
    parser.add_argument('-g', '--gas', type=lambda x: -1 if "air" == x else 1,
                        help="The starting gas for the challenge. If air then standard GC, "
                             "if o2 then it switches from o2->air", default="o2")
    parser.add_argument('--skipstart', type=int, help="Skip Runs at Start",
                        default=0)
    return parser





def main():
    p = init_argparse()
    args = p.parse_args()

    data_folder = args.input
    prefix = args.prefix

    if prefix is None:
        prefix = ""
        raise RuntimeWarning("Warning: You should usually set a prefix for this analysis.")

    for file in sorted(glob.glob(join(data_folder, "*.hdf5")), key=sort_key):
        data = PAData.from_hdf5(file, "r+")
        scan_name = data.get_scan_name().lower()
        prefices = prefix.lower().split(",")
        if not any([x in scan_name for x in prefices]):
            continue
        data.clear_dso2()
        print(scan_name, file)

        scan_unmixed = data.get_scan_unmixed()
        if not scan_unmixed:
            print("Skipped", file, "- no unmixing.")
            continue

        analyser = DCEAnalyser(args.window, args.display, args.sigma, args.buffer)
        try:
            for recon in list(scan_unmixed.keys()):
                unmixed = scan_unmixed[recon]
                run_pipeline(analyser, unmixed, data, -1, True)
        except RuntimeError:
            print("---- SCAN NOT PROCESSED (no reference region) ---")
        data.file.close()



if __name__ == "__main__":
    main()