3-prediction.py

from Packages import *
st.set_page_config(page_title="NIRS Utils", page_icon=":goat:", layout="wide")
from Modules import *
from utils.DATA_HANDLING import *
# HTML pour le bandeau "CEFE - CNRS"
# bandeau_html = """
# <div style="width: 100%; background-color: #4682B4; padding: 10px; margin-bottom: 10px;">
#   <h1 style="text-align: center; color: white;">CEFE - CNRS</h1>
# </div>
# """
# # Injecter le code HTML du bandeau
# st.markdown(bandeau_html, unsafe_allow_html=True)
add_header()
add_sidebar(pages_folder)

local_css(css_file / "style_model.css")
hash_ = ''
def p_hash(add):
    global hash_
    hash_ = hash_data(hash_+str(add))
    return hash_

dirpath = Path('Report/out/model')
if dirpath.exists() and dirpath.is_dir():
    rmtree(dirpath)

if 'Predict' not in st.session_state:
    st.session_state['Predict'] = False
# ####################################  Methods ##############################################
# empty temp figures
def delete_files(keep):
    supp = []
    # Walk through the directory
    for root, dirs, files in os.walk('Report/', topdown=False):
        for file in files:
            if file != 'logo_cefe.png' and not any(file.endswith(ext) for ext in keep):
                os.remove(os.path.join(root, file))
###################################################################
                
st.header("Prediction making using a previously developed model")
c1, c2 = st.columns([2, 1])
c1.image("./images/prediction making.png", use_column_width=True)
pred_data = DataFrame





def preparespecdf(df):
    other = df.select_dtypes(exclude = 'float')
    spec = df.select_dtypes(include='float')
    if other.shape[1] > 0:
        rownames = other.iloc[:,0]
        spec.index = rownames
    else:
        rownames = [str(i) for i in range(df.shape[0])]
    if spec.shape[1]<60:
        spec = DataFrame
    return spec, other, rownames

def check_exist(var):
    out = var in globals()
    return out

with c2:
    zip = st.file_uploader("Load your zip file:", type = ['.zip'], help=" :mushroom: select a csv matrix with samples as rows and lambdas as columns")
    if not zip:
        st.info('Info: Insert your zip file above!')
    
    disable1 = False if zip else True
    new_data = st.file_uploader("Load NIRS Data for prediction making:", type = ['csv', 'dx'], help=" :mushroom: select a csv matrix with samples as rows and lambdas as columns", disabled=disable1)
    if not disable1 :
        info1 = st.info('Info: Insert your NIRS data file above!')

    if zip:
        @st.cache_data
        def tempdir(prefix, dir):
            with TemporaryDirectory( prefix= prefix, dir= dir ) as temp_dir:# create a temp directory
                tempdirname = os.path.split(temp_dir)[1]
            return tempdirname
        
        temp_dir = tempdir(prefix = "pred_temp", dir = "./temp")
        # Open and extract the zip file
        from zipfile import ZipFile
        with ZipFile(zip, 'r') as zip_ref:
            zip_ref.extractall(temp_dir)
    
            
        def find_pkl_files(root_dir):
            # List to store the paths of .pkl files
            pkl_files = []

            # Walk through the directory
            for dirpath, dirnames, filenames in os.walk(root_dir):
                for filename in filenames:
                    # Check if the file has a .pkl extension
                    if filename.endswith('.pkl'):
                        # Construct the full file path
                        file_path = os.path.join(dirpath, filename)
                        pkl_files.append(file_path)
            return pkl_files
        pkl = find_pkl_files(root_dir=temp_dir)

        system_file = [path for path in pkl if 'file_system' in path]
        if len(system_file) ==1 :
            with open(system_file[0], 'rb') as fi:
                system_data = load(fi)

if new_data:
        info1.empty()

with c2:
    if new_data:
        p_hash(new_data.name)
        test = new_data.name.split('.')[-1]
        export_name = 'Pred of'
        export_name += new_data.name[:new_data.name.find('.')]

        match test:
            case 'csv':
                qsep = st.radio("Select csv separator : " , options = [';', ','], key = 2, horizontal = True)
                qhdr = st.radio("indexes column in csv? : " , options = ['yes', 'no'], key = 3, horizontal = True)
                col = 0 if qhdr == 'yes' else None
                p_hash([qsep,qhdr])

                df = read_csv(new_data, sep=qsep, header= col, decimal=".")
                pred_data, cat, rownames =  preparespecdf(df)

            case "dx":
                with NamedTemporaryFile(delete=False, suffix=".dx") as tmp:
                    tmp.write(new_data.read())
                    tmp_path = tmp.name
                    with open(tmp.name, 'r') as dd:
                        dxdata = new_data.read()
                        p_hash(str(dxdata)+str(new_data.name))

                    ## load and parse the temp dx file
                    @st.cache_data
                    def dx_loader(change):
                        chem_data, spectra, meta_data, _ = read_dx(file =  tmp_path)
                        return chem_data, spectra, meta_data, _
                    chem_data, spectra, meta_data, _ = dx_loader(change = hash_)
                    st.success("The data have been loaded successfully", icon="✅")
                    if chem_data.to_numpy().shape[1]>0:
                        yname = st.selectbox('Select target', options=chem_data.columns)
                        measured = chem_data.loc[:,yname] == 0
                        y = chem_data.loc[:,yname].loc[measured]
                        pred_data = spectra.loc[measured]
                    
                    else:
                        pred_data = spectra
                os.unlink(tmp_path)

# Load parameters
st.subheader("I - Spectral data preprocessing & visualization", divider='blue')
# try:
if not pred_data.empty:# Load the model with joblib
    @st.cache_data
    def preprocess_spectra(data, change):
        # M4.write(ProcessLookupError)
        
        if system_data['spec-preprocessing']['normalization'] == 'Snv':
            x1 = Snv(data)
            norm = 'Standard Normal Variate'
        else:
            norm = 'No Normalization was applied'
            x1 = data
        x2 = savgol_filter(x1,
                            window_length = int(system_data['spec-preprocessing']['SavGol(polyorder,window_length,deriv)'][1]),
                            polyorder = int(system_data['spec-preprocessing']['SavGol(polyorder,window_length,deriv)'][0]),
                            deriv = int(system_data['spec-preprocessing']['SavGol(polyorder,window_length,deriv)'][2]),
                                delta=1.0, axis=-1, mode="interp", cval=0.0)
        preprocessed = DataFrame(x2, index = data.index, columns = data.columns)
        return norm, preprocessed
    norm, preprocessed = preprocess_spectra(pred_data, change= hash_)

                        # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ #
    # @st.cache_data
    # def specplot_raw(change):
    #     fig2 = plot_spectra(pred_data, xunits = 'lab', yunits = "meta_data.loc[:,'yunits'][0]")
    #     return fig2
    # rawspectraplot = specplot_raw(change = hash_)
    rawspectraplot = plot_spectra(pred_data, xunits = 'Wavelength/Wavenumber', yunits = "Signal intensity")

    c3, c4 = st.columns([2, 1])
    with c3:
        st.write('Raw spectra')
        st.pyplot(rawspectraplot)

        ## plot preprocessed spectra
        if check_exist("preprocessed"):
            # def specplot_prep(change):
            #     fig2 = plot_spectra(preprocessed, xunits = 'lab', yunits = "meta_data.loc[:,'yunits'][0]")
            #     return fig2
            # prepspectraplot = specplot_prep(change = hash_)
            prepspectraplot = plot_spectra(preprocessed, xunits = 'Wavelength/Wavenumber', yunits = "Signal intensity")
            st.write('Preprocessed spectra')
            st.pyplot(prepspectraplot)
    with c4:
        @st.cache_data
        def prep_info(change):
            SG = f'- Savitzky-Golay derivative parameters \n:(Window_length:{system_data['spec-preprocessing']['SavGol(polyorder,window_length,deriv)'][1]};  polynomial order: {system_data['spec-preprocessing']['SavGol(polyorder,window_length,deriv)'][0]};  Derivative order : {system_data['spec-preprocessing']['SavGol(polyorder,window_length,deriv)'][2]})'
            Norm = f'- Spectral Normalization \n: {system_data['spec-preprocessing']['normalization']}'
            return SG, Norm
        SG, Norm = prep_info(change = hash_)
        st.info('The spectra were preprocessed using:\n'+SG+"\n"+Norm)

    ################### Predictions making  ##########################
    st.subheader("II - Prediction making", divider='blue')
    
    disable2 = False if check_exist("pred_data") else True
    pred_button = st.button('Predict', type='primary', disabled= disable2, use_container_width=False)
    if pred_button:st.session_state['Predict'] = True

    if st.session_state['Predict']:
        if  check_exist("pred_data"):# Load the model with joblib
            c5, c6 = st.columns([2, 1])
            with c6:
                model = system_data['model_']
                if system_data['model_type'] in ['PLS','TPE-iPLS']:
                    nvar = system_data['model_'].n_features_in_
                elif system_data['model_type']  =='LW-PLS':
                    nvar = system_data['data']['raw-spectra'].shape[1]


        if check_exist('preprocessed'):
            if isinstance(system_data['selected-wls']['idx'], DataFrame):
                idx = np.concatenate([np.arange(system_data['selected-wls']['idx'].values.reshape((-1,))[2*i],system_data['selected-wls']['idx'].values.reshape((-1,))[2*i+1]+1) for i in range(system_data['selected-wls']['idx'].shape[0])])
            else:
                idx = np.arange(nvar)

            if np.max(idx) <= preprocessed.shape[1]:
                preprocesseddf = preprocessed.iloc[:,idx] ### get predictors
            else:
                st.error("Error: The number of columns in your data does not match the number of columns used to train the model. Please ensure they are the same.")
            

        if check_exist("preprocesseddf"):
            if st.session_state['Predict'] and nvar == preprocesseddf.shape[1]:
            # if nvar == preprocesseddf.shape[1]:
                match system_data['model_type']:
                    case 'PLS'|'TPE-iPLS':
                        try:
                            result = DataFrame(system_data['model_'].predict(preprocesseddf), index = rownames, columns = ['Results'])
                        except:
                            st.error(f'''Error: Length mismatch: the number of samples indices is {len(rownames)}, while the model produced 
                                            {len(model.predict(preprocesseddf))} values. correct the "indexes column in csv?" parameter''')
                    case 'LW-PLS':
                        temp_path = Path('temp/')
                        # export data to csv for Julia train/pred
                        st.write(system_data['data'])
                        # spectra = system_data['data']['raw-spectra'] # without pretreatments
                        spectra = preprocess_spectra(system_data['data']['raw-spectra'], change= hash_)
                        # with pretreatments
                        x_pred = preprocessed
                        y = system_data['data']['target']
                        data_to_work_with = ['spectra', 'y', 'x_pred']
                        spectra_np, y_np, x_pred_np = spectra.to_numpy(), y.to_numpy(), x_pred.to_numpy()
                        # export spectra, y, x_pred to temp folder as csv files
                        for i in data_to_work_with:
                            j = globals()[i]
                            # st.write(j)
                            np.savetxt(temp_path / str(i + ".csv"), j, delimiter=",")
                        # # run Julia Jchemo as subprocess
                        import subprocess
                        subprocess_path = Path("utils/")
                        # subprocess.run([f"{sys.executable}", subprocess_path / "LWPLSR_Call.py"])
                        # # retrieve json results from Julia JChemo
                        # try:
                        #     with open(temp_path / "lwplsr_outputs.json", "r") as outfile:
                        #         Reg_json = json.load(outfile)
                        #         # delete csv files
                        #         for i in data_to_work_with: os.unlink(temp_path / str(i + ".csv"))
                        #     # delete json file after import
                        #     os.unlink(temp_path / "lwplsr_outputs.json")
                        #     os.unlink(temp_path / "lwplsr_preTreatments.json")
                        #     # format result data into Reg object
                        #     pred = ['pred_data_train', 'pred_data_test']### keys of the dict
                        #     for i in range(nb_folds):
                        #         pred.append("CV" + str(i+1)) ### add cv folds keys to pred
                        # except FileNotFoundError as e:
                        #     Reg = None
                        #     for i in data_to_work_with: os.unlink(temp_path / str(i + ".csv"))
                        #
                        # st.write(Reg_json)
                    

            ################################### results display ###################################
        if check_exist("preprocesseddf"):
            if preprocesseddf.shape[1]>1 and check_exist('result'):
                hist = pred_hist(pred=result)
                with c5:
                    st.write('Predicted values distribution')
                    st.pyplot(hist)
                    st.write('Predicted values table')
                    st.dataframe(result.T)
                with c6:
                    st.info('descriptive statistics for the model output')
                    st.write(DataFrame(desc_stats(result)))
                    
            elif pred_button and nvar != preprocesseddf.shape[1]:
                with c6:
                    st.error(f'Error: The model was trained on {nvar} wavelengths, but you provided {preprocessed.shape[1]} wavelengths for prediction. Please ensure they match!')

            ################################# Download results #################################
        if check_exist('result'):
            @st.cache_data(show_spinner =False)
            def preparing_results_for_downloading(change):
                match test:
                    # load csv file
                    case 'csv':
                        df.to_csv('Report/out/dataset/'+ new_data.name, sep = ';', encoding = 'utf-8', mode = 'a')
                    case 'dx':
                        with open('Report/out/dataset/'+new_data.name, 'w') as dd:
                            dd.write(dxdata)

                prepspectraplot.savefig('./Report/out/figures/raw_spectra.png')
                rawspectraplot.savefig('./Report/out/figures/preprocessed_spectra.png')
                hist.savefig('./Report/out/figures/histogram.png')
                result.round(4).to_csv('./Report/out/The analysis result.csv', sep = ";")

                return change
            preparing_results_for_downloading(change = hash_)

            @st.cache_data(show_spinner =False)
            def tempdir(change):
                with  TemporaryDirectory( prefix="results", dir="./Report") as temp_dir:# create a temp directory
                    tempdirname = os.path.split(temp_dir)[1]
                    if len(os.listdir('./Report/out/figures/'))==3:
                        make_archive(base_name="./Report/Results", format="zip", base_dir="out", root_dir = "./Report")# create a zip file
                        move("./Report/Results.zip", f"./Report/{tempdirname}/Results.zip")# put the inside the temp dir
                        with open(f"./Report/{tempdirname}/Results.zip", "rb") as f:
                            zip_data = f.read()
                return tempdirname, zip_data

            date_time = datetime.now().strftime('%y%m%d%H%M')
            try :
                tempdirname, zip_data = tempdir(change = hash_)
                st.download_button(label = 'Download', data = zip_data, file_name = f'Nirs_Workflow_{date_time}_Pred_.zip', mime ="application/zip",
                            args = None, kwargs = None,type = "primary",use_container_width = True)
            except:
                st.write('rtt')
        # except:
        #     c2.error('''Error: Data loading failed. Please check your file. Consider fine-tuning the dialect settings or ensure the file isn't corrupted.''')


else:
    with c2:
        if new_data:
            st.error("Error!:The The data you provided for making predictions doesn't appear to be multivariable.!")