Skip to content

print_and_save_data

create_and_save_dataframes(estimates, bvp_names, bvp_methods, landmark_name_dict, results_folder, print_mean)

Creates and saves dataframes with the provided estimates.

Parameters:

Name Type Description Default
estimates list

List of estimate data.

 required
bvp_names list

List of BVP names.

 required
bvp_methods list

List of BVP methods.

 required
landmark_name_dict dict

Dictionary mapping landmark names.

 required
results_folder str

Folder path to save the dataframes.

 required
print_mean bool

Flag indicating whether to print mean values.

 required

Returns:

Type Description

None
Source code in redesign_pipeline/utils/print_and_save_data.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def create_and_save_dataframes(estimates,bvp_names,bvp_methods,landmark_name_dict,results_folder,print_mean):
    """Creates and saves dataframes with the provided estimates.

    Args:
        estimates (list): List of estimate data.
        bvp_names (list): List of BVP names.
        bvp_methods (list): List of BVP methods.
        landmark_name_dict (dict): Dictionary mapping landmark names.
        results_folder (str): Folder path to save the dataframes.
        print_mean (bool): Flag indicating whether to print mean values.

    Returns:
        None
    """

    header = ['Window num', 'BVP Name', 'BVP Method', 'ROI No', 'ROI Name', 'HR', 'RR', 'SpO2', 'RMSSD', 'SDNN', 'PNN50', 'LF', 'HF']

    df_rows = []
    for estimate in estimates:
        for idx,bvp_name in enumerate(bvp_names):
            for bvp_method in bvp_methods:
                ins_values = {'HR':[], 'RR':[], 'SpO2':[], 'RMSSD':[], 'SDNN':[], 'PNN50':[], 'LF':[], 'HF':[]}
                for lm_i,landmark in landmark_name_dict.items():

                    hr_value = estimate[2][idx][bvp_method][lm_i].item() if isinstance(estimate[2][idx][bvp_method],list) else estimate[2][idx][bvp_method].item()
                    ins_values['HR'].append(hr_value)
                    ins_values['RR'].append(estimate[3][0][bvp_method][0])
                    ins_values['SpO2'].append(float(estimate[1][lm_i]))
                    ins_values['RMSSD'].append(estimate[4][idx][bvp_method][0][0])
                    ins_values['SDNN'].append(estimate[4][idx][bvp_method][0][1])
                    ins_values['PNN50'].append(estimate[4][idx][bvp_method][0][2])
                    ins_values['LF'].append(estimate[4][idx][bvp_method][0][3])
                    ins_values['HF'].append(estimate[4][idx][bvp_method][0][4])

                    df_rows.append((estimate[0],bvp_name,bvp_method,lm_i,landmark,hr_value,
                                    estimate[3][0][bvp_method][lm_i],
                                    float(estimate[1][lm_i]),estimate[4][idx][bvp_method][0][0],estimate[4][idx][bvp_method][0][1],estimate[4][idx][bvp_method][0][2],
                                    estimate[4][idx][bvp_method][0][3],estimate[4][idx][bvp_method][0][4]))

                df_rows.append((estimate[0],bvp_name,bvp_method,'AVG','-',mean(ins_values['HR']),mean(ins_values['RR'])
                                ,mean(ins_values['SpO2']),mean(ins_values['RMSSD']),mean(ins_values['SDNN']),mean(ins_values['PNN50']),mean(ins_values['LF']),mean(ins_values['HF']))) 

                df_rows.append((estimate[0],bvp_name,bvp_method,'MEDIAN','-',median(ins_values['HR']),median(ins_values['RR'])
                                ,median(ins_values['SpO2']),median(ins_values['RMSSD']),median(ins_values['SDNN']),median(ins_values['PNN50']),median(ins_values['LF']),median(ins_values['HF']))) 


    df_results = pd.DataFrame(df_rows, columns=header).round(2) 
    instantaneous_csv = os.path.join(results_folder, 'instantaneous.csv')
    df_results.to_csv(instantaneous_csv)

    if print_mean:
        logger.info(f"Mean values for the {bvp_methods[0]} method and {bvp_names[0]}")
        filtered_df = df_results[(df_results['BVP Method']==bvp_methods[0]) & (df_results['BVP Name']==bvp_names[0]) & (df_results['ROI Name']!='-')][['ROI Name','HR','RR','SpO2','RMSSD','SDNN','PNN50','LF','HF']]
        filtered_df["SpO2"] = pd.to_numeric(filtered_df["SpO2"])
        mean_df = filtered_df.groupby(['ROI Name']).mean(numeric_only=True).astype(float).round(2)
        logger.info(tabulate(mean_df, headers='keys', tablefmt='fancy_grid'))

print_window_results(hr, rr, hrv, bvp_methods, landmark_name_dict, window, selected_lm)

Prints the results for a given window.

Parameters:

Name Type Description Default
hr dict

Dictionary containing HR values.

 required
rr dict

Dictionary containing RR values.

 required
hrv dict

Dictionary containing HRV values.

 required
bvp_methods list

List of BVP methods.

 required
landmark_name_dict dict

Dictionary mapping landmark names.

 required
window int

Window number.

 required
selected_lm str

Selected landmark name.

 required

Returns:

Type Description

None
Source code in redesign_pipeline/utils/print_and_save_data.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def print_window_results(hr,rr,hrv,bvp_methods,landmark_name_dict,window,selected_lm):
    """Prints the results for a given window.

    Args:
        hr (dict): Dictionary containing HR values.
        rr (dict): Dictionary containing RR values.
        hrv (dict): Dictionary containing HRV values.
        bvp_methods (list): List of BVP methods.
        landmark_name_dict (dict): Dictionary mapping landmark names.
        window (int): Window number.
        selected_lm (str): Selected landmark name.

    Returns:
        None
    """

    logger.info(f'For Window No: window')
    for method in bvp_methods:
        logger.info(f'Using {method} BVP method')
        hr_rois =  hr[method][selected_lm] if isinstance(hr[method],list) else hr[method]
        hrv_rois = hrv[method] 
        rr_rois = rr[method]

        landmark = landmark_name_dict[selected_lm]
        logger.info('{}(HR){:.0f} (RR){:.0f} (RMSSD){:.2f} (SDNN){:.2f} (PNN50){:.0f} (LF){:.2f} (HF){:.2f}'.format(landmark, hr_rois, rr_rois[selected_lm], hrv_rois[selected_lm][0], hrv_rois[selected_lm][1], hrv_rois[selected_lm][2], hrv_rois[selected_lm][3], hrv_rois[selected_lm][4]))
        logger.info("")

save_intermediate_data(data_folder, window_count, rgb_list, bvp_named_dict)

Saves intermediate data to files.

Parameters:

Name Type Description Default
data_folder str

Folder path to save the data.

 required
window_count int

Window count.

 required
rgb_list list

RGB data.

 required
bvp_named_dict dict

Dictionary containing BVP data.

 required

Returns:

Type Description

None
Source code in redesign_pipeline/utils/print_and_save_data.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def save_intermediate_data(data_folder,window_count,rgb_list,bvp_named_dict):
    """Saves intermediate data to files.

    Args:
        data_folder (str): Folder path to save the data.
        window_count (int): Window count.
        rgb_list (list): RGB data.
        bvp_named_dict (dict): Dictionary containing BVP data.

    Returns:
        None
    """
    rgb_folder = os.path.join(data_folder,'rgb')
    bvp_folder = os.path.join(data_folder,'bvp')
    os.makedirs(rgb_folder,exist_ok=True)
    os.makedirs(bvp_folder,exist_ok=True)

    rgb_file = open(os.path.join(rgb_folder, f'rgb_window_{window_count}.pckl'),'wb')

    pickle.dump(rgb_list,rgb_file)

    rgb_file.close()

    for bvp_name in bvp_named_dict:
        bvp_folder = os.path.join(bvp_folder,bvp_name)
        os.makedirs(bvp_folder,exist_ok=True)

        bvp_file = open(os.path.join(bvp_folder, f'{bvp_name}_window_{window_count}.pckl'),'wb')

        pickle.dump(bvp_named_dict[bvp_name],bvp_file)

        bvp_file.close()